from __future__ import annotations

import csv
import io
import uuid
from datetime import UTC, date, datetime
from typing import Any, Literal

from app.core.safe_logging import get_safe_logger

logger = get_safe_logger(__name__)

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

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    check_approval_submit_rate_limit,
    get_current_user,
    get_real_ip,
    require_browser_session,
)
from app.core.dependencies import (
    require_active_employee_for_mutations as _leaver_runtime_gate,
)
from app.core.exceptions import (
    AppException,
    ForbiddenError,
    NotFoundError,
    ProdCloudLockedError,
    ValidationError,
)
from app.core.rbac import (
    CLOUD_ADMIN_ROLE,
    EMPLOYEE_ROLE,
    GLOBAL_ADMIN_ROLE,
    can_manage_cloud,
    can_manage_cloud_scope_lifecycle,
    can_read_cloud,
    can_view_cloud_resources,
    can_view_finance,
    get_subordinate_person_ids,
    has_role,
)
from app.core.response import paginated_response, success_response
from app.domains.cmdb import cert_service, dedicated_line_service
from app.domains.cmdb.credential_crypto import (
    CredentialDecryptionError,
    decrypt_fields,
)
from app.domains.cmdb.providers.base import (
    DnsRecordWrite,
    ProviderCredentials,
    get_provider_class,
)
from app.domains.cmdb.scheduled_jobs_provisioning import (
    resolve_sync_capabilities,
)
from app.domains.cmdb.schemas import (
    BillingDailyPoint,
    BillingMonthlyResponse,
    BillingOverview,
    BillingProviderBreakdown,
    BillingSummary,
    CdnDomainAnnotation,
    CdnDomainResponse,
    CertificateAnnotation,
    CertificateCreate,
    CertificateResponse,
    DedicatedLineCreate,
    DedicatedLineResponse,
    DedicatedLineUpdate,
    DnsRecordBatchOwnerTransfer,
    DnsRecordBatchOwnerUpdate,
    DnsRecordCreate,
    DnsRecordRequestCreate,
    DnsRecordResponse,
    DnsRecordUpdate,
    DomainAnnotation,
    DomainResponse,
    DomainSelectableForDns,
    ForceSweepRequest,
    IamUserCleanupRequest,
    IamUserOwnerUpdate,
    IamUserResponse,
    LoadBalancerAnnotation,
    LoadBalancerBatchAnnotation,
    LoadBalancerResponse,
    ObjectStorageAnnotation,
    ObjectStorageResponse,
    ServerAnnotation,
    ServerBatchAnnotation,
    ServerBatchOwnerTransfer,
    ServerResponse,
    SyncJobResponse,
    SyncTriggerRequest,
)
from app.domains.cmdb.service import (
    CmdbService,
    _active_cloud_integration_ids,
    _esc,
    resolve_inherited_server_owners,
)
from app.domains.cmdb.sync import run_sync, run_sync_with_lock
from app.domains.cmdb.visibility import (
    OwnerVisibility,
    is_server_visible,
    resolve_owner_visibility,
    visibility_clause,
)
from app.integrations.models import Integration

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


async def _load_cloud_integration(
    db: AsyncSession, integration_id: str,
) -> Integration:
    """Load an ``Integration`` row and assert ``kind='cloud'``.

    Raises 404 for unknown id or non-cloud integrations — per-kind
    enforcement keeps the CMDB router from acting on, say, a HRIS or
    chat-approval integration.
    """
    row = await db.get(Integration, uuid.UUID(integration_id))
    if row is None or row.kind != "cloud":
        raise NotFoundError("CloudIntegration", integration_id)
    return row


async def _resolve_owner_labels(
    db: AsyncSession, person_ids: list[str | None],
) -> dict[str, str]:
    """Batch-load identity_persons and return {person_id: "Display · email"}.

    Used by the CSV export endpoints since migration 042 dropped the
    free-text ``owner_raw`` column — owner display text is now derived
    from the owner_person_id FK in one query.
    """
    clean_ids = [pid for pid in person_ids if pid]
    if not clean_ids:
        return {}
    from sqlalchemy import select as _select

    from app.identity.person_models import IdentityPerson

    rows = await db.execute(
        _select(IdentityPerson).where(IdentityPerson.id.in_(clean_ids)),
    )
    people = list(rows.scalars().all())
    labels: dict[str, str] = {}
    for p in people:
        display = p.display_name or ""
        email = p.work_email or ""
        if display and email:
            labels[str(p.id)] = f"{display} · {email}"
        else:
            labels[str(p.id)] = display or email or str(p.id)
    return labels


async def _resolve_owner_meta(
    db: AsyncSession, person_ids: list[str | None],
) -> dict[str, dict[str, str | None]]:
    """Batch-load {person_id: {"employee_no", "employment_status"}}.

    Companion to ``_resolve_owner_labels`` — feeds the leaver-visibility
    fields (``owner_employee_no`` / ``owner_employment_status``) so the
    frontend PersonTag can strike-through + 「已离职」-tag a departed owner.
    Kept as a separate IN-query (rather than fattening the label resolver
    that ~25 string-only call sites depend on); identity_persons lookups
    are cheap and batched per endpoint.
    """
    clean_ids = [pid for pid in person_ids if pid]
    if not clean_ids:
        return {}
    from sqlalchemy import select as _select

    from app.identity.person_models import IdentityPerson

    rows = await db.execute(
        _select(
            IdentityPerson.id,
            IdentityPerson.employee_no,
            IdentityPerson.employment_status,
        ).where(IdentityPerson.id.in_(clean_ids)),
    )
    meta: dict[str, dict[str, str | None]] = {}
    for pid, emp_no, status in rows.all():
        meta[str(pid)] = {
            "employee_no": emp_no,
            "employment_status": status,
        }
    return meta


async def _resolve_person_labels_for_rows(
    db: AsyncSession, rows: list[Any],
) -> dict[str, str]:
    """Combined owner + modifier label resolver.

    Migration 173 introduced ``last_modified_by_person_id`` on five
    resource tables. The list endpoints now render two person FKs
    per row (owner + last modifier); resolving them in two separate
    batch queries doubles the round-trip count and the connection
    pressure. Take the union here and let callers index by id.
    """
    pids: list[str] = []
    for r in rows:
        owner = getattr(r, "owner_person_id", None)
        modifier = getattr(r, "last_modified_by_person_id", None)
        if owner:
            pids.append(str(owner))
        if modifier:
            pids.append(str(modifier))
    return await _resolve_owner_labels(db, pids)


async def _resolve_labels_and_owner_meta(
    db: AsyncSession, rows: list[Any],
) -> tuple[dict[str, str], dict[str, dict[str, str | None]]]:
    """``_resolve_person_labels_for_rows`` + owner leaver-meta in one go.

    Returns ``(labels, owner_meta)``; pass both to ``_augment_response``
    so departed owners surface 工号 + 在职状态 to PersonTag."""
    labels = await _resolve_person_labels_for_rows(db, rows)
    owner_meta = await _resolve_owner_meta(
        db, [getattr(r, "owner_person_id", None) for r in rows],
    )
    return labels, owner_meta


def _augment_response(
    item: dict,
    row: Any,
    labels: dict[str, str],
    owner_meta: dict[str, dict[str, str | None]] | None = None,
) -> dict:
    """Populate owner_label and last_modified_by_label on a serialized
    row from the per-call label cache built by
    ``_resolve_person_labels_for_rows``.

    When ``owner_meta`` (from ``_resolve_owner_meta``) is supplied, also
    populate ``owner_employee_no`` / ``owner_employment_status`` so the
    frontend PersonTag can mark a departed owner."""
    owner = getattr(row, "owner_person_id", None)
    if owner:
        item["owner_label"] = labels.get(str(owner))
        if owner_meta is not None:
            m = owner_meta.get(str(owner))
            if m:
                item["owner_employee_no"] = m["employee_no"]
                item["owner_employment_status"] = m["employment_status"]
    modifier = getattr(row, "last_modified_by_person_id", None)
    if modifier:
        item["last_modified_by_label"] = labels.get(str(modifier))
    return item


async def _self_visibility(db: AsyncSession, user: AuthUser) -> OwnerVisibility:
    """Self-service visibility for a non-admin reader: servers they (or
    their transitive reports) own at server level, plus every server
    attributed to a cloud subscription they (or their reports) own. The
    single scoping primitive shared by the servers list/detail and the
    efficiency surface — see ``visibility.py``."""
    subs = await get_subordinate_person_ids(db, str(user.id))
    person_ids = [str(user.id), *subs] if subs else [str(user.id)]
    return await resolve_owner_visibility(db, person_ids)


async def _augment_inherited_owners(
    db: AsyncSession, items: list[dict], servers: list[Any],
) -> None:
    """For serialized server rows with no explicit owner, resolve+attach the
    owner inherited from the cloud subscription (live, never stored). Mutates
    ``items`` in place. Batched: one resolver pass + one label/meta lookup for
    the inherited owners."""
    inherited = await resolve_inherited_server_owners(db, servers)
    if not inherited:
        return
    pids = [v["person_id"] for v in inherited.values() if v.get("person_id")]
    labels = await _resolve_owner_labels(db, pids)
    meta = await _resolve_owner_meta(db, pids)
    for item in items:
        info = inherited.get(str(item.get("id")))
        if not info:
            continue
        pid = info["person_id"]
        item["owner_inherited"] = True
        item["effective_owner_person_id"] = pid
        item["effective_owner_label"] = labels.get(str(pid))
        m = meta.get(str(pid))
        if m:
            item["effective_owner_employee_no"] = m["employee_no"]
            item["effective_owner_employment_status"] = m["employment_status"]
        item["inherited_from_scope_id"] = info["scope_id"]
        item["inherited_from_scope_name"] = info["scope_name"]


async def _resolve_integration_meta(
    db: AsyncSession, integration_ids: list[str | None],
) -> dict[str, dict[str, str | None]]:
    """Batch-resolve ``integration_id`` → ``{provider, name}``.

    Populated onto ServerResponse so the frontend can:
      * gate the per-row "更多 → 开机/关机/重启/删除" action menu
        per-provider (only ucloud / aliyun / aws / alicloud_intl
        implement cloud lifecycle methods today; the rest surface 501)
      * show a cloud-vendor pill in the detail page without a second
        round-trip

    One IN-query regardless of N rows. Orphan rows (integration row
    deleted) are left without the keys; the frontend handles
    ``provider == null`` as "actions unavailable".
    """
    clean_ids = {iid for iid in integration_ids if iid}
    if not clean_ids:
        return {}
    result = await db.execute(
        select(Integration).where(Integration.id.in_(clean_ids))
    )
    out: dict[str, dict[str, str | None]] = {}
    for integ in result.scalars().all():
        out[str(integ.id)] = {
            "provider": integ.provider,
            "integration_name": integ.name,
        }
    return out


def _attach_integration_meta(
    item: dict, row: Any, meta: dict[str, dict[str, str | None]],
) -> dict:
    """Inject ``provider`` + ``integration_name`` onto a serialized row.

    Pair with ``_augment_response`` in the server list / get endpoints.
    Mutates + returns ``item`` for fluent chaining.
    """
    iid = str(getattr(row, "integration_id", "") or "")
    info = meta.get(iid)
    if info:
        item["provider"] = info.get("provider")
        item["integration_name"] = info.get("integration_name")
    return item


_CSV_INJECTION_PREFIXES = ("=", "+", "-", "@", "\t", "\r")


def _csv_safe(value: Any) -> Any:
    """Defuse CSV-formula injection (`=cmd|...!A1`, etc.).

    A cell starting with one of the formula trigger characters in a
    spreadsheet program (Excel, Numbers, LibreOffice) gets evaluated
    as a formula on open, allowing remote-payload exfiltration or
    local command execution. Prefixing those values with a single
    quote tells the spreadsheet to treat them as literal text. Apply
    to every user-controlled string written into an export.
    """
    if isinstance(value, str) and value.startswith(_CSV_INJECTION_PREFIXES):
        return "'" + value
    return value


def _csv_safe_row(row: list[Any]) -> list[Any]:
    return [_csv_safe(c) for c in row]


async def _notify_owner_of_transfer(
    db: AsyncSession,
    *,
    new_owner_person_id: str,
    actor: AuthUser,
    resource_kind: str,  # "dns" | "server"
    resource_count: int,
    reason: str | None,
    trace_id: str,
) -> None:
    """Dispatch a Feishu + email notification to the new owner after a
    self-service handoff.

    Best-effort: the notification outbox persists the row before any
    channel is tried, so a transient Feishu / SMTP outage produces a
    retryable row, not a silent drop. The caller is expected to wrap
    this in a try/except — failure NEVER blocks the transfer commit.
    """
    from app.identity.person_models import IdentityPerson
    from app.notifications.service import NotificationEnvelope, NotificationService

    recipient = await db.get(IdentityPerson, uuid.UUID(new_owner_person_id))
    if recipient is None or not recipient.work_email:
        return
    actor_person = await db.get(IdentityPerson, actor.id)
    actor_label = (
        getattr(actor_person, "display_name", None)
        or getattr(actor_person, "work_email", None)
        or "Helix 用户"
    )
    lang_raw = (recipient.preferred_language or "").lower()
    if lang_raw.startswith("en"):
        lang = "en"
    elif lang_raw.startswith("ja"):
        lang = "ja"
    else:
        lang = "zh"  # default for unspecified / cn / zh-CN
    # Resource-kind → deep link + i18n copy table. Adding a new
    # resource type means a new row, not another if-branch.
    deep_link = (
        "/my/cloud/dns" if resource_kind == "dns" else "/my/cloud/servers"
    )
    related_resource_type = (
        "cmdb_dns_record" if resource_kind == "dns" else "cmdb_server"
    )
    copy: dict[tuple[str, str], tuple[str, str, str]] = {
        ("dns", "zh"): (
            f"[已转移] {actor_label} 把 {resource_count} 条 DNS 记录交给你",
            f"{actor_label} 把 {resource_count} 条 DNS 解析记录"
            "转移给了你。你现在是这些记录的负责人,保鲜确认 / 修改 / "
            "删除申请都在你名下。请在 Helix → 我的资源 → DNS 解析 中查看。",
            "转移备注",
        ),
        ("dns", "en"): (
            f"[Transfer] {actor_label} transferred "
            f"{resource_count} DNS record(s) to you",
            f"{actor_label} has handed off {resource_count} DNS "
            "record(s) to you. You are now the owner; recertify "
            "and self-service requests run under your name. View "
            "the list in Helix → My Resources → DNS records.",
            "Reason",
        ),
        ("dns", "ja"): (
            f"[譲渡] {actor_label} が {resource_count} 件の DNS レコードを譲渡しました",
            f"{actor_label} が {resource_count} 件の DNS レコードを"
            "あなたに譲渡しました。あなたが新しい担当者です。"
            "Helix → 私のリソース → DNS レコードで一覧を確認してください。",
            "譲渡メモ",
        ),
        ("server", "zh"): (
            f"[已转移] {actor_label} 把 {resource_count} 台服务器交给你",
            f"{actor_label} 把 {resource_count} 台云服务器"
            "转移给了你。你现在是这些服务器的负责人,半年盘点和离职"
            "清理都会指向你。请在 Helix → 我的资源 → 云服务器 中查看。",
            "转移备注",
        ),
        ("server", "en"): (
            f"[Transfer] {actor_label} transferred "
            f"{resource_count} server(s) to you",
            f"{actor_label} has handed off {resource_count} "
            "cloud server(s) to you. You are now the owner; "
            "half-yearly confirmations and offboarding sweeps will "
            "target you. View the list in Helix → My Resources → Servers.",
            "Reason",
        ),
        ("server", "ja"): (
            f"[譲渡] {actor_label} が {resource_count} 台のサーバを譲渡しました",
            f"{actor_label} が {resource_count} 台のクラウドサーバを"
            "あなたに譲渡しました。半年棚卸と退職整理はあなたに向きます。"
            "Helix → 私のリソース → サーバで一覧を確認してください。",
            "譲渡メモ",
        ),
    }
    title, body, reason_label = copy[(resource_kind, lang)]
    if reason:
        body = body + f"\n\n{reason_label}: {reason}"
    notify = NotificationService(db)
    await notify.dispatch(
        NotificationEnvelope(
            recipient=recipient,
            title=title,
            body_text=body,
            channels=["feishu", "email"],
            url=deep_link,
            related_resource_type=related_resource_type,
            # No single related_resource_id — a batch transfer touches
            # many rows. Outbox dedup is per-recipient, so multiple
            # transfers within a backoff window appear as separate
            # rows, which is what we want.
            related_resource_id=None,
        ),
        trace_id=trace_id,
    )


def _can_submit_self_service_dns(role_names) -> bool:  # type: ignore[no-untyped-def]
    """Gate for ``POST /dns-record-requests`` and
    ``POST /dns-records/recertify``.

    These endpoints mutate state (drive cloud DNS writes via approval,
    re-stamp ``last_owner_review_at``). The default authenticated user
    is too permissive — pure read-only privileged roles (``auditor``,
    ``management``) should not be able to submit. Allow the canonical
    mutators: ``employee`` (every human submitting their own change),
    ``global_admin`` (admins driving on someone's behalf), and
    ``cloud_admin`` (network-domain authority since 2026-05-20 — they
    operate DNS as part of their write scope, same as global_admin).
    """
    return has_role(role_names, EMPLOYEE_ROLE, GLOBAL_ADMIN_ROLE, CLOUD_ADMIN_ROLE)


def _scrub_audit_patch(record_type: str | None, patch: dict[str, Any]) -> dict[str, Any]:
    """Apply ``_audit_dns_value`` redaction to any ``value`` key in a
    PATCH dict before writing into ``audit_logs.detail``. Idempotent.
    """
    if "value" in patch and patch["value"] is not None:
        patch = dict(patch)
        patch["value"] = _audit_dns_value(record_type, patch["value"])
    return patch


def _audit_dns_value(record_type: str | None, value: str | None) -> str | None:
    """Redact DNS record values that may carry secrets before persisting
    into audit_logs.detail.

    TXT records carry domain-ownership tokens, ACME-DNS-01 challenges,
    SaaS verification secrets, and occasional misuse as token storage.
    audit_logs is readable by every ``security_auditor``; persisting
    raw TXT values here means a future audit export carries those
    secrets in plaintext. Hash-prefix preserves compare-by-equality
    forensic queries without disclosing the original value. Non-TXT
    types are not redacted — A / AAAA / CNAME / MX / SRV / CAA / NS
    values are operationally interesting and not secret.
    """
    if value is None:
        return None
    if (record_type or "").upper() != "TXT":
        return value
    import hashlib
    digest = hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()
    return f"[TXT redacted; sha256:{digest[:16]} length={len(value)}]"


def _csv_response(rows: list[dict], filename: str) -> StreamingResponse:
    """Build a CSV StreamingResponse from a list of flat dicts.

    Prepends a UTF-8 BOM so Excel-on-Windows opens non-ASCII content
    (Chinese server / domain names) without falling back to Latin-1 /
    GBK autodetect. Same pattern as ``cost_service.stream_facts_csv``.
    """
    if not rows:
        content = ""
    else:
        buf = io.StringIO()
        writer = csv.DictWriter(buf, fieldnames=list(rows[0].keys()))
        writer.writeheader()
        for r in rows:
            writer.writerow({k: _csv_safe(v) for k, v in r.items()})
        content = buf.getvalue()

    body = ("" + content).encode("utf-8")
    return StreamingResponse(
        iter([body]),
        media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


# ---------------------------------------------------------------------------
# Stats
# ---------------------------------------------------------------------------


@router.get("/stats")
async def get_stats(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    stats = await CmdbService.get_resource_stats(db)
    return success_response(data=stats, request=request)


@router.get("/cloud-tenants")
async def list_cloud_tenants(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Thin list of cloud integrations for UI tenant filters — id,
    name, slug, provider, enabled. No secrets. Employees can call this
    so the ``<CloudTenantFilter>`` works on ``/my/cloud/*`` pages too.

    Distinct from ``/integrations`` which is admin-only and returns
    full config including approval flows.
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    rows = (
        await db.execute(
            select(
                Integration.id, Integration.name, Integration.slug,
                Integration.provider, Integration.enabled,
            )
            .where(Integration.kind == "cloud")
            .order_by(Integration.enabled.desc(), Integration.name.asc())
        )
    ).all()
    data = [
        {
            "id": str(r.id), "name": r.name, "slug": r.slug,
            "provider": r.provider, "enabled": bool(r.enabled),
        }
        for r in rows
    ]
    return success_response(data=data, request=request)


@router.get("/analytics")
async def get_analytics(
    request: Request,
    billing_months: int = 12,
    integration_id: str | None = None,
    department: str | None = Query(None, max_length=128),
    measure: str | None = Query(
        default=None,
        description=(
            "成本口径:``list``(默认,业务折前成本/研发口径,按月聚合自 "
            "cmdb_cloud_billing_daily)或 ``invoice``(实付,读 "
            "cmdb_billing_monthly,仅 finance / cloud_admin / "
            "global_admin)。只影响 ``billing_trend`` 与 "
            "``per_tenant_summary[].latest_month_cost_usd``。"
        ),
    ),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Fleet-wide analytics: counts-by-provider/region/status/…, domain
    expiry buckets, monthly billing trend, per-tenant summary cards.
    Backs both /admin/cloud/overview and /admin/network/overview.

    When ``integration_id`` is set, every aggregation is scoped to that
    ONE cloud tenant (drill-down mode). ``None`` = all tenants, with
    per-tenant breakdowns exposed in the ``by_integration`` arrays.

    When ``department`` is set, every *owner-attributable* resource
    aggregation (servers, DNS records, domains, load balancers, CDN, IAM
    — any model carrying ``owner_person_id``) is narrowed to resources
    owned by people in that department. Resources with no owner column
    (object storage today) cannot be attributed and report zero under the
    filter; the monthly billing trend is tenant-level and stays
    department-agnostic (per-department cost lives in the billing tab's
    own drill-down).

    口径路由（2026-07 保密红线补口）：``can_read_cloud`` 的受众含
    **auditor / management**（两者都在保密圈外），而 ``billing_trend`` +
    ``per_tenant_summary[].latest_month_cost_usd`` 此前裸读
    ``cmdb_billing_monthly``（纯实付）。同一批人在 ``/billing/monthly``
    已经拿到 (月 × integration × provider) 粒度的 **list** 数 ——
    两个面一相除就是每云每月的公司折扣率。默认改为 list（按月聚合日表），
    ``invoice`` 仅保密圈可请求（无权限 → 403，绝不静默降级）。
    """
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.cost_view import resolve_measure

    try:
        resolved_measure = resolve_measure(measure, user.role_names)
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc
    if resolved_measure is None:
        raise ForbiddenError()
    dept = (department.strip() if department else None) or None
    data = await CmdbService.get_analytics(
        db,
        owner_person_id=None,
        billing_months=billing_months,
        integration_id=integration_id,
        department=dept,
        measure=resolved_measure,
    )
    return success_response(data=data, request=request)


@router.get("/analytics/my")
async def get_my_analytics(
    request: Request,
    integration_id: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Employee-scoped analytics: same shape as /analytics but every
    aggregation is filtered to rows the current user owns
    (``owner_person_id = user.id``) plus rows owned by their direct +
    transitive subordinates if they are a manager. Mirrors the scope
    used by the corresponding list endpoints (``/cmdb/domains``,
    ``/cmdb/dns-records``, ``/cmdb/cdn-domains``) so the dashboard
    numbers reconcile with what the list pages show.

    Billing trend + per-tenant cards are omitted because bills are
    tenant-level, not per-owner.

    ``integration_id`` narrows further to one cloud tenant — useful if
    an employee has resources across multiple clouds.
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    # Manager-scope expansion — same logic as ``list_servers`` /
    # ``list_domains`` / ``list_dns_records`` / ``list_cdn_domains``.
    # Pre-fix this endpoint always passed only ``user.id``, so a
    # manager's overview undercounted vs the list pages. See B6 in the
    # pre-launch audit.
    owner_person_id: str | None = str(user.id)
    owner_person_ids: list[str] | None = None
    server_visibility: OwnerVisibility | None = None
    if not can_read_cloud(user.role_names):
        subs = await get_subordinate_person_ids(db, str(user.id))
        if subs:
            owner_person_ids = [str(user.id), *subs]
            owner_person_id = None
        # Servers additionally count owned-subscription attribution, so
        # the overview's server number reconciles with the servers list
        # and the efficiency page (shared visibility contract).
        server_visibility = await _self_visibility(db, user)
    data = await CmdbService.get_analytics(
        db,
        owner_person_id=owner_person_id,
        owner_person_ids=owner_person_ids,
        owner_email=user.work_email,
        integration_id=integration_id,
        server_visibility=server_visibility,
    )
    return success_response(data=data, request=request)


# ---------------------------------------------------------------------------
# Sync trigger — now driven by integration_id
# ---------------------------------------------------------------------------


@router.post("/sync/{integration_id}/trigger", status_code=202)
async def trigger_sync(
    integration_id: uuid.UUID,
    body: SyncTriggerRequest,
    request: Request,
    background_tasks: BackgroundTasks,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    integration = await _load_cloud_integration(db, str(integration_id))
    if not integration.enabled:
        raise NotFoundError("CloudIntegration", str(integration_id))
    user_id_string = str(user.id)
    job = await CmdbService.create_sync_job(
        db,
        integration_id=str(integration.id),
        resource_types=body.resource_types,
        triggered_by_user_id=user_id_string,
    )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.sync.trigger",
        status="success",
        resource_type="integration",
        resource_id=str(integration.id),
        detail={
            "integration_id": str(integration.id),
            "resource_types": body.resource_types,
        },
    )
    # Pre-launch hardening 2026-05-06: COMMIT the sync_job + audit row
    # BEFORE scheduling the background task. Previously the
    # ``add_task`` was registered while the request session was still
    # uncommitted; FastAPI runs background tasks AFTER the request
    # completes, but if the request commit failed (FK error, deadlock,
    # disconnect) the job row never landed while the background task
    # ran anyway and silently no-op'd because ``db.get(CmdbSyncJob, id)``
    # returned None. Committing first guarantees the background worker
    # sees a real job row.
    await db.commit()
    background_tasks.add_task(
        run_sync,
        str(integration.id),
        str(job.id),
        body.resource_types,
        user_id_string,
    )
    response = SyncJobResponse.model_validate(job)
    return success_response(data=response.model_dump(), request=request)


# ---------------------------------------------------------------------------
# Sync Jobs
# ---------------------------------------------------------------------------


@router.post("/sync/{integration_id}/trigger-all", status_code=202)
async def trigger_sync_all(
    integration_id: uuid.UUID,
    request: Request,
    background_tasks: BackgroundTasks,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Trigger one manual sync for every capability supported by this
    integration. Backs the "一键全量同步" button on the cloud
    integrations list page.

    This is NOT the same as iterating the scheduled_jobs rows — it
    queues a *single* CmdbSyncJob that covers all caps at once so the
    admin sees one job row in the UI. Per-capability scheduling
    continues to fire on its own cron independently.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    integration = await _load_cloud_integration(db, str(integration_id))
    if not integration.enabled:
        raise NotFoundError("CloudIntegration", str(integration_id))
    effective = resolve_sync_capabilities(integration)
    if not effective:
        raise NotFoundError("ProviderCapabilities", integration.provider)
    user_id_string = str(user.id)
    job = await CmdbService.create_sync_job(
        db,
        integration_id=str(integration.id),
        resource_types=effective,
        triggered_by_user_id=user_id_string,
    )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.sync.trigger_all",
        status="success",
        resource_type="integration",
        resource_id=str(integration.id),
        detail={
            "integration_id": str(integration.id),
            "capabilities": effective,
        },
    )
    # Pre-launch hardening 2026-05-06: see trigger_sync — commit
    # before scheduling the background task so the worker doesn't
    # start with a not-yet-persisted job row.
    await db.commit()
    background_tasks.add_task(
        run_sync,
        str(integration.id),
        str(job.id),
        effective,
        user_id_string,
    )
    response = SyncJobResponse.model_validate(job)
    return success_response(data=response.model_dump(), request=request)


@router.post("/sync/{integration_id}/force-sweep", status_code=202)
async def force_resource_sweep_endpoint(
    integration_id: uuid.UUID,
    body: ForceSweepRequest,
    request: Request,
    background_tasks: BackgroundTasks,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_browser_session),
):
    """Admin-confirmed override of the cmdb resource-sync safety floor.

    Why this exists: ``_safe_sweep`` refuses to soft-delete when the
    provider returns either zero rows or fewer than 95% of the existing
    row count — defending against a partial driver failure silently
    mass-deleting real production rows. When the admin has intentionally
    deleted N users in the cloud console (or a region has truly emptied
    out) the floor blocks every subsequent cron with the same error,
    the failure_notifier keeps re-paging on every tick, and ``ack`` only
    silences the alert without fixing the underlying state. This
    endpoint is the explicit escape valve: re-run sync for the chosen
    capabilities with ``force_sweep=True`` so the missing rows are
    finally soft-deleted and the cron returns to green.

    Behaviour:
      * Requires ``can_manage_cloud``. 403 otherwise.
      * ``resource_types`` is required and must be non-empty — the
        admin picks which capabilities to force-sweep. No wildcard.
      * Queues the sync as a FastAPI BackgroundTask and returns 202
        with the initial job row immediately. The frontend polls
        ``GET /cmdb/sync-jobs/{id}`` to watch for terminal status and
        render the ``capability_results.<cap>.forced_sweep`` metadata
        (which guard tripped, prior ``existing`` / ``upserted`` counts).
        The background task acquires the ``cloud_sync:<integration>``
        distributed lock internally — same isolation against the scope /
        grants paths — and surfaces lock contention as a ``status=
        'failed'`` job row with a clear retry message (instead of the
        pre-async 409). Bug 2026-05-18: when this endpoint ran sync
        synchronously, a multi-project GCP fan-out exceeded nginx's
        default 60s ``proxy_read_timeout`` and the user saw "后端返回
        错误，但没有详细消息" — the empty error envelope produced by a
        504 from the proxy.
      * Writes a ``cmdb.sync.force_sweep_override`` audit row
        attributing the decision to the admin user *at request time*
        (the decision is made now, regardless of how the background
        sync ultimately completes). Soft-deletes themselves are
        reversible (``deleted_at=NULL``); the audit trail is the
        forever record of WHO chose to bypass.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    integration = await _load_cloud_integration(db, str(integration_id))
    if not integration.enabled:
        raise NotFoundError("CloudIntegration", str(integration_id))
    # Reject capabilities the provider doesn't support — fail-fast so
    # the admin gets a clear error instead of a sync with every entry
    # landing on "skipped: not supported".
    supported = set(resolve_sync_capabilities(integration))
    requested = list(dict.fromkeys(body.resource_types))  # de-dup, preserve order
    unsupported = [c for c in requested if c not in supported]
    if unsupported:
        raise ValidationError(
            f"Capabilities not supported by provider "
            f"{integration.provider!r}: {unsupported}",
        )

    integ_id_str = str(integration.id)
    user_id_string = str(user.id)
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))

    job = await CmdbService.create_sync_job(
        db,
        integration_id=integ_id_str,
        resource_types=requested,
        triggered_by_user_id=user_id_string,
        triggered_by="force_sweep",
    )
    job_id_str = str(job.id)
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.sync.force_sweep_override",
        status="success",
        resource_type="integration",
        resource_id=integ_id_str,
        detail={
            "integration_id": integ_id_str,
            "provider": integration.provider,
            "resource_types": requested,
            "job_id": job_id_str,
            "reason": body.reason,
        },
    )
    # Pre-launch hardening 2026-05-06: see ``trigger_sync`` — commit
    # the sync_job + audit row BEFORE scheduling the BG task so the
    # background worker reads a persisted job row, not a not-yet-
    # committed one.
    await db.commit()

    background_tasks.add_task(
        run_sync_with_lock,
        integ_id_str,
        job_id_str,
        requested,
        user_id_string,
        force_sweep=True,
    )

    response = SyncJobResponse.model_validate(job)
    return success_response(data=response.model_dump(), request=request)


@router.get("/sync-jobs")
async def list_sync_jobs(
    request: Request,
    integration_id: str | None = None,
    limit: int = Query(default=20, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    jobs = await CmdbService.list_sync_jobs(db, integration_id=integration_id, limit=limit)
    data = [SyncJobResponse.model_validate(j).model_dump() for j in jobs]
    return success_response(data=data, request=request)


@router.get("/sync-jobs/{job_id}")
async def get_sync_job(
    job_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbSyncJob
    result = await db.execute(
        select(CmdbSyncJob).where(CmdbSyncJob.id == job_id)
    )
    job = result.scalar_one_or_none()
    if not job:
        raise NotFoundError("SyncJob", str(job_id))
    return success_response(data=SyncJobResponse.model_validate(job).model_dump(), request=request)


@router.get("/integrations/{integration_id}/sync-capability-status")
async def integration_sync_capability_status(
    integration_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Diagnostic roll-up: latest sync outcome per capability for one
    cloud account (status / counts / reason / partial-failure count /
    last-run time), including capabilities that have never synced. Lets
    an admin answer "why did servers sync but IAM fail?" without reading
    the raw capability_results JSON, and drives the per-capability re-run
    button (which posts the existing /sync/{id}/trigger with one
    resource_type)."""
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.sync_status import build_capability_status
    integration = await _load_cloud_integration(db, str(integration_id))
    return success_response(
        data=await build_capability_status(db, integration),
        request=request,
    )


# ---------------------------------------------------------------------------
# Recycle bin (admin escape hatch — list + restore soft-deleted rows)
#
# Declared BEFORE every ``/{resource_type}/{id}`` endpoint so FastAPI's
# left-to-right path resolution doesn't try to parse the literal
# ``recycle-bin`` as a resource UUID — same trick as
# ``/cmdb/dns-zones`` / ``/cmdb/domains/selectable-for-dns`` /
# ``GET /assets/recycle-bin``.
#
# Rationale: the legacy-CMDB → UCloud-sync collision incident
# (2026-04-24) silently soft-deleted 116 UCloud servers because of an
# instance_id format mismatch; admins had no UI to find / restore them.
# Alembic 212 brings the swept rows back; these endpoints make the
# recovery surface permanent + visible so future drift is recoverable.
# ---------------------------------------------------------------------------


# Whitelist of recycle-bin–enabled resource types. Drives the
# resource_type validation at both list + restore endpoints; mirrored
# in MCP tools (`_CMDB_RECYCLE_BIN_TYPES`) and in the per-route model
# map at restore-time. ``object_storage`` / ``iam_user`` are
# intentionally absent: those are sync-managed without admin
# annotation (no owner_person_id / app / remark to preserve across
# a delete/restore cycle), so a recycle bin there has no concrete
# recovery story yet.
_RECYCLE_BIN_TYPES: frozenset[str] = frozenset({
    "server", "domain", "dns_record", "load_balancer", "cdn_domain",
    # ``certificate`` carries admin annotations (owner_person_id /
    # environment / remark) worth preserving across a delete/restore
    # cycle — same bucket as server/domain — and the cert sync upsert
    # deliberately no longer auto-resurrects a swept/retired cloud cert
    # (``deleted_at`` omitted from on_duplicate_key_update). Without a
    # recycle-bin entry that left a swept cert permanently unrecoverable,
    # contradicting the "recovered by admin Restore" promise in the cert
    # upsert comment.
    "certificate",
})


@router.get("/recycle-bin/{resource_type}")
async def list_recycle_bin(
    resource_type: str,
    request: Request,
    integration_id: str | None = None,
    search: str | None = None,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List soft-deleted CMDB rows of one resource type, newest-deleted
    first. Admin-only — no employee read-down path because a deleted
    cloud row by definition has no live owner relationship to demote
    against.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    if resource_type not in _RECYCLE_BIN_TYPES:
        raise NotFoundError("RecycleBinResource", resource_type)
    from app.domains.cmdb.models import (
        CmdbCdnDomain as _Cdn,  # noqa: PLC0415
    )
    from app.domains.cmdb.models import (
        CmdbCertificate as _Cert,  # noqa: PLC0415
    )
    from app.domains.cmdb.models import (
        CmdbDnsRecord as _Dns,  # noqa: PLC0415
    )
    from app.domains.cmdb.models import (
        CmdbDomain as _Dom,  # noqa: PLC0415
    )
    from app.domains.cmdb.models import (
        CmdbLoadBalancer as _Lb,  # noqa: PLC0415
    )
    from app.domains.cmdb.models import (
        CmdbServer as _Srv,  # noqa: PLC0415
    )
    if resource_type == "server":
        rows, total = await CmdbService.list_deleted_servers(
            db, integration_id=integration_id, search=search,
            page=page, page_size=page_size,
        )
        response_cls = ServerResponse
        _model_cls = _Srv  # noqa: F841 — selected for parity / docs
    elif resource_type == "domain":
        rows, total = await CmdbService.list_deleted_domains(
            db, integration_id=integration_id, search=search,
            page=page, page_size=page_size,
        )
        response_cls = DomainResponse
        _model_cls = _Dom  # noqa: F841
    elif resource_type == "dns_record":
        rows, total = await CmdbService.list_deleted_dns_records(
            db, integration_id=integration_id, search=search,
            page=page, page_size=page_size,
        )
        response_cls = DnsRecordResponse
        _model_cls = _Dns  # noqa: F841
    elif resource_type == "load_balancer":
        rows, total = await CmdbService.list_deleted_load_balancers(
            db, integration_id=integration_id, search=search,
            page=page, page_size=page_size,
        )
        response_cls = LoadBalancerResponse
        _model_cls = _Lb  # noqa: F841
    elif resource_type == "cdn_domain":
        rows, total = await CmdbService.list_deleted_cdn_domains(
            db, integration_id=integration_id, search=search,
            page=page, page_size=page_size,
        )
        response_cls = CdnDomainResponse
        _model_cls = _Cdn  # noqa: F841
    else:  # certificate
        rows, total = await CmdbService.list_deleted_certificates(
            db, integration_id=integration_id, search=search,
            page=page, page_size=page_size,
        )
        response_cls = CertificateResponse
        _model_cls = _Cert  # noqa: F841
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, rows)
    data = [
        _augment_response(response_cls.model_validate(r).model_dump(), r, labels, owner_meta)
        for r in rows
    ]
    return paginated_response(
        data=data, total=total, page=page, page_size=page_size, request=request,
    )


@router.post("/recycle-bin/{resource_type}/{resource_id}/restore")
async def restore_recycle_bin_row(
    resource_type: str,
    resource_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Clear ``deleted_at`` on a soft-deleted CMDB row. Admin-only.

    409 if the row is not deleted (idempotency guard against
    double-click / stale recycle-bin list). Audit row's
    ``resource_type`` matches the existing audit conventions
    (``cmdb_server`` / ``cmdb_domain`` / etc.) so the restored row's
    timeline picks it up.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import (
        CmdbCdnDomain,  # noqa: PLC0415
        CmdbCertificate,  # noqa: PLC0415
        CmdbDnsRecord,  # noqa: PLC0415
        CmdbDomain,  # noqa: PLC0415
        CmdbLoadBalancer,  # noqa: PLC0415
        CmdbServer,  # noqa: PLC0415
    )
    model_map: dict[str, tuple[Any, str]] = {
        "server":        (CmdbServer,       "cmdb_server"),
        "domain":        (CmdbDomain,       "cmdb_domain"),
        "dns_record":    (CmdbDnsRecord,    "cmdb_dns_record"),
        "load_balancer": (CmdbLoadBalancer, "cmdb_load_balancer"),
        "cdn_domain":    (CmdbCdnDomain,    "cmdb_cdn_domain"),
        "certificate":   (CmdbCertificate,  "cmdb_certificate"),
    }
    if resource_type not in model_map:
        raise NotFoundError("RecycleBinResource", resource_type)
    model_cls, audit_resource_type = model_map[resource_type]
    result = await CmdbService.restore_resource(
        db,
        model=model_cls,
        resource_id=str(resource_id),
        actor_person_id=str(user.id),
    )
    if result is None:
        raise NotFoundError(audit_resource_type, str(resource_id))
    row, forensics = result
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.resource.restore",
        status="success",
        resource_type=audit_resource_type,
        resource_id=str(resource_id),
        detail={"restored": True, **forensics},
    )
    return success_response(
        data={"id": str(row.id), "resource_type": audit_resource_type, "restored": True},
        request=request,
    )


# ---------------------------------------------------------------------------
# Servers
# ---------------------------------------------------------------------------


async def _resolve_provider_integration_ids(
    db: AsyncSession, provider: str | None,
) -> list[str] | None:
    """Resolve a cloud provider name to the list of integration ids.

    Returns ``None`` when ``provider`` is unset (caller should not
    pass an ``integration_ids`` filter at all). Returns an empty list
    when the provider exists in our enum but no enabled tenant uses
    it — service.list_servers short-circuits to zero rows in that
    case (correct: "no aliyun tenant configured" = "no aliyun rows").
    """
    if not provider:
        return None
    from app.integrations.models import Integration as _Integration
    q = select(_Integration.id).where(_Integration.provider == provider)
    result = await db.execute(q)
    return [str(r) for r in result.scalars().all()]


@router.get("/servers")
async def list_servers(
    request: Request,
    integration_id: str | None = None,
    provider: str | None = None,
    status: str | None = None,
    region: str | None = None,
    zone: str | None = None,
    project_id: str | None = None,
    instance_type: str | None = None,
    os_type: str | None = None,
    vcpu_min: int | None = Query(None, ge=1),
    vcpu_max: int | None = Query(None, ge=1),
    age_min_days: int | None = Query(None, ge=0),
    age_max_days: int | None = Query(None, ge=0),
    search: str | None = None,
    owner_person_id: str | None = None,
    page: int = 1,
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    visibility: OwnerVisibility | None = None
    if not can_read_cloud(user.role_names):
        # Self-service scope: own + team (manager expansion) servers at
        # server level, plus servers attributed to cloud subscriptions
        # they own — same visibility contract as the efficiency page,
        # so a machine never shows on one and 404s on the other.
        visibility = await _self_visibility(db, user)
        owner_person_id = None
    integration_ids = await _resolve_provider_integration_ids(db, provider)
    servers, total = await CmdbService.list_servers(
        db,
        integration_id=integration_id,
        integration_ids=integration_ids,
        status=status,
        region=region,
        zone=zone,
        project_id=project_id,
        instance_type=instance_type,
        os_type=os_type,
        vcpu_min=vcpu_min,
        vcpu_max=vcpu_max,
        age_min_days=age_min_days,
        age_max_days=age_max_days,
        search=search,
        owner_person_id=owner_person_id,
        visibility=visibility,
        page=page,
        page_size=page_size,
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, servers)
    integ_meta = await _resolve_integration_meta(
        db, [str(s.integration_id) for s in servers],
    )
    data = [
        _attach_integration_meta(
            _augment_response(
                ServerResponse.model_validate(s).model_dump(), s, labels, owner_meta,
            ),
            s, integ_meta,
        )
        for s in servers
    ]
    await _augment_inherited_owners(db, data, servers)
    return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request)


@router.get("/servers/export")
async def export_servers(
    request: Request,
    integration_id: str | None = None,
    provider: str | None = None,
    status: str | None = None,
    region: str | None = None,
    os_type: str | None = None,
    vcpu_min: int | None = Query(None, ge=1),
    vcpu_max: int | None = Query(None, ge=1),
    age_min_days: int | None = Query(None, ge=0),
    age_max_days: int | None = Query(None, ge=0),
    search: str | None = None,
    owner_person_id: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Export all matching servers as CSV (up to 5000 rows)."""
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    visibility: OwnerVisibility | None = None
    if not can_read_cloud(user.role_names):
        visibility = await _self_visibility(db, user)
        owner_person_id = None
    integration_ids = await _resolve_provider_integration_ids(db, provider)
    servers, _ = await CmdbService.list_servers(
        db,
        integration_id=integration_id,
        integration_ids=integration_ids,
        status=status,
        region=region,
        os_type=os_type,
        vcpu_min=vcpu_min,
        vcpu_max=vcpu_max,
        age_min_days=age_min_days,
        age_max_days=age_max_days,
        search=search,
        owner_person_id=owner_person_id,
        visibility=visibility,
        page=1,
        page_size=5000,
    )
    # Batch-resolve owner_person_id to work_email / display_name for
    # the "owner" CSV column. One query regardless of N servers;
    # owner_raw column was dropped in migration 042.
    owner_labels = await _resolve_owner_labels(
        db, [s.owner_person_id for s in servers if s.owner_person_id],
    )
    rows = []
    for s in servers:
        rows.append({
            "instance_id": s.instance_id,
            "name": s.name or "",
            "status": s.status,
            "region": s.region or "",
            "zone": s.zone or "",
            "vcpu": s.vcpu or "",
            "memory_mb": s.memory_mb or "",
            "os_type": s.os_type or "",
            "public_ips": "|".join(s.public_ips or []),
            "private_ips": "|".join(s.private_ips or []),
            "owner": owner_labels.get(s.owner_person_id, "") if s.owner_person_id else "",
            "last_synced_at": str(s.last_synced_at)[:19],
        })
    return _csv_response(rows, "cmdb_servers.csv")


@router.get("/servers/filter-options")
async def get_server_filter_options(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Distinct provider / region / os_type values for the filter dropdowns.

    Scoped to whatever the caller can see: employees get only their
    own + subordinates' rows; cloud-admins get the global enabled
    tenant set. This guarantees the dropdown never shows a value
    that would return zero rows after RBAC filtering — important UX,
    otherwise users see an "aliyun" option and get no results.

    Returns ``{providers, regions, os_types, instance_type_tiers}``.
    ``instance_type_tiers`` is a static set keyed off ``vcpu`` ranges
    that the frontend uses to populate the "规格档位" dropdown — server
    side decides the buckets so the labels stay consistent.
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from sqlalchemy import distinct

    from app.domains.cmdb.models import CmdbServer
    from app.integrations.models import Integration as _Integration

    # Collect the scoping predicates once, reuse them across the three
    # distinct-value lookups. Mirrors list_servers' RBAC branch.
    scope_filters: list[Any] = [CmdbServer.deleted_at.is_(None)]
    if not can_read_cloud(user.role_names):
        scope_filters.append(
            visibility_clause(await _self_visibility(db, user))
        )
    else:
        active = await _active_cloud_integration_ids(db)
        if not active:
            return success_response(request=request, data={
                "providers": [],
                "regions": [],
                "os_types": [],
                "instance_type_tiers": _INSTANCE_TYPE_TIERS,
            })
        scope_filters.append(CmdbServer.integration_id.in_(active))

    region_rows = await db.execute(
        select(distinct(CmdbServer.region)).where(
            *scope_filters, CmdbServer.region.is_not(None),
        )
    )
    regions = sorted([r for r in region_rows.scalars().all() if r])

    os_rows = await db.execute(
        select(distinct(CmdbServer.os_type)).where(
            *scope_filters, CmdbServer.os_type.is_not(None),
        )
    )
    os_types = sorted([o for o in os_rows.scalars().all() if o])

    integ_rows = await db.execute(
        select(distinct(CmdbServer.integration_id)).where(*scope_filters)
    )
    integ_ids = [str(i) for i in integ_rows.scalars().all() if i]
    providers: list[str] = []
    if integ_ids:
        prov_rows = await db.execute(
            select(distinct(_Integration.provider)).where(_Integration.id.in_(integ_ids))
        )
        providers = sorted([p for p in prov_rows.scalars().all() if p])

    return success_response(request=request, data={
            "providers": providers,
            "regions": regions,
            "os_types": os_types,
            "instance_type_tiers": _INSTANCE_TYPE_TIERS,
        },
    )


# Static spec-tier buckets exposed by /servers/filter-options. The
# frontend renders ``label`` and submits ``vcpu_min`` / ``vcpu_max``
# (kept None when unset to mean "open-ended"). Buckets cover the
# 1C-128C range the prod inventory actually has — adjust if larger
# VMs become common.
_INSTANCE_TYPE_TIERS = [
    {"value": "small", "label_zh": "小 (1-2C)", "label_en": "Small (1-2 vCPU)",  "vcpu_min": 1,  "vcpu_max": 2},
    {"value": "medium", "label_zh": "中 (4-8C)", "label_en": "Medium (4-8 vCPU)", "vcpu_min": 4, "vcpu_max": 8},
    {"value": "large",  "label_zh": "大 (16-32C)", "label_en": "Large (16-32 vCPU)", "vcpu_min": 16, "vcpu_max": 32},
    {"value": "xlarge", "label_zh": "超大 (≥64C)", "label_en": "XLarge (≥64 vCPU)", "vcpu_min": 64, "vcpu_max": None},
]


@router.get("/servers/overview-tops")
async def get_server_overview_tops(
    request: Request,
    integration_id: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Top-N panels for the cloud-servers overview tab.

    Returns two leaderboards:
      * ``owner_top`` — owner_person_id with the most servers attached.
        Useful to spot accounts that have accumulated too much resource
        weight (potential candidate for governance follow-up).
      * ``long_idle_top`` — stopped servers ordered by ``last_modified_at``
        ascending (oldest stopped first). Acts as a "candidate for
        decommission" list; sync writes ``last_modified_at`` on any
        state change, so a server that's been stopped for ages will
        have an old timestamp here.

    Scoping mirrors ``list_servers`` (RBAC + active-tenant filter).
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from sqlalchemy import func as _func

    from app.domains.cmdb.models import CmdbServer

    scope_filters: list[Any] = [CmdbServer.deleted_at.is_(None)]
    if not can_read_cloud(user.role_names):
        scope_filters.append(
            visibility_clause(await _self_visibility(db, user))
        )
    else:
        active = await _active_cloud_integration_ids(db)
        if not active:
            return success_response(
                request=request, data={"owner_top": [], "long_idle_top": []},
            )
        scope_filters.append(CmdbServer.integration_id.in_(active))

    # Selected-tenant narrowing: when the overview tab has a tenant
    # picked, the sibling analytics call already scopes to it — the
    # Top-N leaderboards must follow suit or they'd stay global while
    # the charts next to them are filtered (KPI-vs-chart drift).
    if integration_id:
        scope_filters.append(CmdbServer.integration_id == integration_id)

    # Owner top — group by owner_person_id, count, sort desc, top 10.
    owner_rows = await db.execute(
        select(
            CmdbServer.owner_person_id,
            _func.count(CmdbServer.id).label("cnt"),
        )
        .where(*scope_filters, CmdbServer.owner_person_id.is_not(None))
        .group_by(CmdbServer.owner_person_id)
        .order_by(_func.count(CmdbServer.id).desc())
        .limit(10)
    )
    owner_rows_list = owner_rows.all()
    owner_ids_for_label = [str(r[0]) for r in owner_rows_list if r[0]]
    label_map = await _resolve_owner_labels(db, owner_ids_for_label)
    owner_meta = await _resolve_owner_meta(db, owner_ids_for_label)
    owner_top = [
        {
            "owner_person_id": str(row[0]) if row[0] else None,
            "owner_label": label_map.get(str(row[0])) if row[0] else None,
            "owner_employee_no": (
                owner_meta.get(str(row[0]), {}).get("employee_no") if row[0] else None
            ),
            "owner_employment_status": (
                owner_meta.get(str(row[0]), {}).get("employment_status") if row[0] else None
            ),
            "count": int(row[1]),
        }
        for row in owner_rows_list
    ]

    # Long-idle top — stopped servers ordered by oldest last_modified_at
    # (proxy for "stopped longest"; sync rewrites this field on every
    # state change). last_modified_at can be null on rows the legacy
    # importer landed without modifier metadata — those sort LAST via
    # the ``IS NULL`` sort-key trick (MySQL / TiDB don't support the
    # explicit ``NULLS LAST`` syntax; same pattern list_domains uses).
    long_rows = await db.execute(
        select(CmdbServer)
        .where(*scope_filters, CmdbServer.status == "stopped")
        .order_by(
            CmdbServer.last_modified_at.is_(None),
            CmdbServer.last_modified_at.asc(),
        )
        .limit(10)
    )
    long_servers = list(long_rows.scalars().all())
    long_labels = await _resolve_owner_labels(
        db, [s.owner_person_id for s in long_servers if s.owner_person_id],
    )
    long_owner_meta = await _resolve_owner_meta(
        db, [s.owner_person_id for s in long_servers if s.owner_person_id],
    )
    long_integ_meta = await _resolve_integration_meta(
        db, [str(s.integration_id) for s in long_servers],
    )
    now_ts = datetime.now(UTC)
    long_idle_top = []
    for s in long_servers:
        last_change = s.last_modified_at or s.first_seen_at
        idle_days = None
        if last_change is not None:
            ref = last_change
            if ref.tzinfo is None:
                ref = ref.replace(tzinfo=UTC)
            idle_days = max(0, (now_ts - ref).days)
        meta = long_integ_meta.get(str(s.integration_id), {}) if long_integ_meta else {}
        long_idle_top.append({
            "id": str(s.id),
            "instance_id": s.instance_id,
            "name": s.name,
            "region": s.region,
            "status": s.status,
            "owner_person_id": s.owner_person_id,
            "owner_label": long_labels.get(s.owner_person_id) if s.owner_person_id else None,
            "owner_employee_no": (
                long_owner_meta.get(s.owner_person_id, {}).get("employee_no")
                if s.owner_person_id else None
            ),
            "owner_employment_status": (
                long_owner_meta.get(s.owner_person_id, {}).get("employment_status")
                if s.owner_person_id else None
            ),
            "provider": meta.get("provider"),
            "first_seen_at": s.first_seen_at.isoformat() if s.first_seen_at else None,
            "last_modified_at": s.last_modified_at.isoformat() if s.last_modified_at else None,
            "idle_days": idle_days,
        })

    return success_response(
        data={"owner_top": owner_top, "long_idle_top": long_idle_top},
        request=request,
    )


@router.get("/servers/{server_id}")
async def get_server(
    server_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbServer
    result = await db.execute(
        select(CmdbServer).where(
            CmdbServer.id == server_id,
            CmdbServer.deleted_at.is_(None),
        )
    )
    server = result.scalar_one_or_none()
    if not server:
        raise NotFoundError("Server", str(server_id))
    # Per-row authorization mirrors the list endpoint: owner OR manager
    # OR owned-subscription attribution OR admin. Otherwise an employee
    # can iterate UUIDs and pull every server one by one even though
    # /servers list demotes. Same visibility contract as the list, so a
    # row reachable from any list/efficiency page never 404s here.
    if not can_read_cloud(user.role_names):
        vis = await _self_visibility(db, user)
        if not await is_server_visible(db, str(server_id), vis):
            raise NotFoundError("Server", str(server_id))
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [server])
    integ_meta = await _resolve_integration_meta(
        db, [str(server.integration_id)],
    )
    item = _attach_integration_meta(
        _augment_response(
            ServerResponse.model_validate(server).model_dump(), server, labels, owner_meta,
        ),
        server, integ_meta,
    )
    await _augment_inherited_owners(db, [item], [server])
    return success_response(data=item, request=request)


@router.patch("/servers/{server_id}")
async def annotate_server(
    server_id: uuid.UUID,
    body: ServerAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    server = await CmdbService.annotate_server(
        db, str(server_id), body, actor_person_id=str(user.id),
    )
    if not server:
        raise NotFoundError("Server", str(server_id))
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.server.annotate",
        status="success",
        resource_type="cmdb_server",
        resource_id=str(server_id),
        detail=body.model_dump(exclude_unset=True, mode="json"),
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [server])
    integ_meta = await _resolve_integration_meta(
        db, [str(server.integration_id)],
    )
    item = _attach_integration_meta(
        _augment_response(
            ServerResponse.model_validate(server).model_dump(), server, labels, owner_meta,
        ),
        server, integ_meta,
    )
    return success_response(data=item, request=request)


@router.post("/servers/batch-owner-transfer")
async def batch_transfer_server_owner(
    body: ServerBatchOwnerTransfer,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Self-service cloud-server owner handoff.

    Mirrors ``POST /dns-records/batch-owner-transfer`` over
    ``cmdb_servers``: caller must own every selected server (admin
    bypass), recipient must be an active employee, no approval, new
    owner is notified via the outbox.
    """
    # Same role gate as the DNS variant — employee + global_admin +
    # cloud_admin. ``can_view_cloud_resources`` would be too narrow
    # (auditors / management would slip through); ``can_manage_cloud``
    # too tight (excludes ``employee`` self-service).
    if not _can_submit_self_service_dns(user.role_names):
        raise ForbiddenError()
    is_admin = can_manage_cloud(user.role_names)
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    summary = await CmdbService.transfer_owned_servers(
        db,
        server_ids=[str(sid) for sid in body.server_ids],
        new_owner_person_id=str(body.new_owner_person_id),
        caller_person_id=str(user.id),
        caller_is_admin=is_admin,
        reason=body.reason,
        trace_id=trace_id,
    )
    if summary["not_owned"] and not is_admin:
        raise ForbiddenError(
            f"{summary['not_owned']} server(s) in the batch are not "
            "owned by you. Refresh the list and try again."
        )
    # Refuse to hand off never-sync-confirmed rows (may no longer exist in
    # the cloud — the 2026-07 Oncall_bot dangling-transfer incident). If
    # nothing else transferred, tell the caller why instead of a silent
    # no-op; a partial batch still reports the skipped count in ``summary``.
    if summary.get("unconfirmed") and not summary["transferred"]:
        raise ValidationError(
            f"{summary['unconfirmed']} server(s) can no longer be "
            "transferred — they are pending sync reconciliation and may no "
            "longer exist in the cloud. Refresh the list and try again."
        )
    transferred_ids = summary.pop("_transferred_ids", [])
    audit = AuditService(db)
    # See the parallel comment in batch_transfer_dns_record_owner —
    # resource_id is String(128) so CSV-joining 50 UUIDs truncates.
    first_id = transferred_ids[0] if transferred_ids else ""
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.server.self_service_transfer",
        status="success",
        resource_type="cmdb_server",
        resource_id=first_id,
        detail={
            "summary": summary,
            "new_owner_person_id": str(body.new_owner_person_id),
            "server_ids_count": len(body.server_ids),
            "transferred_ids": transferred_ids,
            "reason": body.reason,
        },
    )
    if transferred_ids:
        try:
            await _notify_owner_of_transfer(
                db,
                new_owner_person_id=str(body.new_owner_person_id),
                actor=user,
                resource_kind="server",
                resource_count=len(transferred_ids),
                reason=body.reason,
                trace_id=trace_id,
            )
        except Exception:  # noqa: BLE001
            logger.exception(
                "Server self-service transfer: notify dispatch failed "
                "caller=%s new_owner=%s count=%d",
                user.id, body.new_owner_person_id, len(transferred_ids),
            )
    return success_response(data=summary, request=request)


@router.post("/servers/batch-annotate")
async def batch_annotate_servers(
    body: ServerBatchAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Apply the same annotation to multiple servers in one round-trip.

    Mirrors the legacy CMDB's ``PATCH /servers/update`` with
    ``server_ids[]`` semantics. Admin-only; intended for bulk owner
    reassignment after a cloud account is freshly imported.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    summary = await CmdbService.batch_annotate_servers(
        db,
        [str(sid) for sid in body.server_ids],
        body.annotation,
        actor_person_id=str(user.id),
    )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.server.batch_annotate",
        status="success",
        resource_type="cmdb_server",
        resource_id=",".join(str(s) for s in body.server_ids[:50]),
        detail={
            "summary": summary,
            "annotation": body.annotation.model_dump(exclude_unset=True, mode="json"),
            "server_ids_count": len(body.server_ids),
        },
    )
    return success_response(data=summary, request=request)


# ---------------------------------------------------------------------------
# Server lifecycle (power on / off / restart / destroy)  ─── PR #559
# ---------------------------------------------------------------------------
#
# Replicates the legacy CMDB 开机 / 关机 / 重启 / 删除 buttons. Each
# endpoint:
#   1. loads the ``cmdb_servers`` row (404 on missing / soft-deleted)
#   2. instantiates the live cloud provider via
#      ``_load_provider_for_dns`` — name is historical, the helper
#      itself is provider-agnostic (loads integration, decrypts
#      credentials, returns a CloudProvider instance)
#   3. calls the matching ``power_on``/``power_off``/``restart_server``/
#      ``destroy_server`` method on the provider
#   4. updates the DB row's ``status`` to the optimistic transitional
#      value (``starting`` / ``stopping`` / ``rebooting`` / ``terminated``)
#      — the next ``cmdb.servers`` sync cron reconciles into the
#      authoritative cloud-side state
#   5. writes an audit_logs row with the operator, action, integration
#      + instance reference, and outcome (success / capability_unsupported)
#
# RBAC: ``can_manage_cloud`` (it_admin / cloud_admin). Reads of cloud
# resources stay on ``can_view_cloud_resources`` — mutating cloud state
# is a separate, narrower role.
#
# Idempotency: the providers themselves catch "instance already in
# target state" responses and return success (see e.g.
# ``UCloudProvider._lifecycle_call`` / ``AliyunProvider._lifecycle_call``).
# A double-click results in a single audit row + 200, not 502.
#
# Provider coverage (2026-05-28):
#   * UCloud, Aliyun, AWS — full implementation
#   * AliCloud Intl — inherits Aliyun's lifecycle methods
#   * Azure, GCP, Tencent, KSyun — no impl; surface 501 with the
#     provider name so the admin knows to use the cloud console


_LIFECYCLE_TRANSITIONAL_STATUS: dict[str, str] = {
    "power_on": "starting",
    "power_off": "stopping",
    "restart": "rebooting",
    "destroy": "terminated",
}


async def _execute_server_lifecycle(
    action: str,
    server_id: uuid.UUID,
    request: Request,
    user: AuthUser,
    db: AsyncSession,
) -> dict[str, Any]:
    """Common execution path for the 4 server lifecycle endpoints.

    ``action`` is the canonical verb (``power_on`` / ``power_off`` /
    ``restart`` / ``destroy``) which maps both to a provider method
    and a transitional DB status.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbServer
    from app.domains.cmdb.providers.exceptions import (
        CapabilityNotSupported,
        ProviderAuthError,
        ProviderTransientError,
        redact_provider_error,
    )
    # Lock the row for the duration of the lifecycle op. ``destroy`` is
    # irreversible (TerminateInstances); without this, two concurrent
    # POSTs (double-click / retry / MCP+UI) both pass the
    # ``deleted_at IS NULL`` read and both fire the terminate. The lock
    # serialises them — the second waits, re-reads after the first commits
    # the soft-delete, no longer matches ``deleted_at IS NULL``, and gets
    # a clean NotFound instead of a second cloud terminate.
    result = await db.execute(
        select(CmdbServer).where(
            CmdbServer.id == server_id,
            CmdbServer.deleted_at.is_(None),
        ).with_for_update()
    )
    server = result.scalar_one_or_none()
    if not server:
        raise NotFoundError("Server", str(server_id))

    integration_id_str = str(server.integration_id)

    # Production kill-switch. The same global prod-write lock that guards
    # per-scope cloud writes must also block destructive server lifecycle
    # ops — destroy is the single highest-blast-radius provider call we make
    # (TerminateInstances), and the lock exists precisely to prevent
    # accidental prod mutation during an incident lockdown. CmdbServer has no
    # ``environment`` of its own; it inherits the environment of its cloud
    # account, which lives on CmdbCloudScope (shared ``integration_id``). We
    # treat the server as prod if its integration has ANY active prod scope —
    # an integration is one connector / one cloud account, so this is precise
    # in the common single-env case and fail-SAFE (over-protect) in the rare
    # mixed case. ``power_on`` is recovery (non-destructive) and stays allowed.
    if action in ("destroy", "power_off", "restart"):
        from app.domains.cmdb.models import (
            PROD_ENVIRONMENTS,
            CmdbCloudScope,
        )
        from app.domains.cmdb.prod_write_lock import (
            LOCKED_MESSAGE,
            prod_cloud_writes_enabled,
        )
        has_prod_scope = (await db.execute(
            select(CmdbCloudScope.id).where(
                CmdbCloudScope.integration_id == integration_id_str,
                CmdbCloudScope.environment.in_(tuple(PROD_ENVIRONMENTS)),
                CmdbCloudScope.deleted_at.is_(None),
            ).limit(1)
        )).first() is not None
        if has_prod_scope and not await prod_cloud_writes_enabled(db):
            raise ProdCloudLockedError(LOCKED_MESSAGE)

    instance_id = server.instance_id
    region = server.region
    project_id = server.project_id

    provider = await _load_provider_for_dns(db, integration_id_str)
    method_name = {
        "power_on": "power_on",
        "power_off": "power_off",
        "restart": "restart_server",
        "destroy": "destroy_server",
    }[action]
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    try:
        method = getattr(provider, method_name)
        await method(
            instance_id=instance_id,
            region=region,
            project_id=project_id,
        )
    except CapabilityNotSupported as exc:
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=user.id,
            action=f"cmdb.server.{action}",
            status="failure",
            resource_type="cmdb_server",
            resource_id=str(server_id),
            detail={
                "reason": "capability_not_supported",
                "provider_message": str(exc)[:300],
                "integration_id": integration_id_str,
                "instance_id": instance_id,
            },
        )
        await db.commit()
        raise AppException(
            code="CLOUD_LIFECYCLE_NOT_SUPPORTED",
            message=str(exc),
            status_code=501,
        ) from exc
    except ProviderAuthError as exc:
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=user.id,
            action=f"cmdb.server.{action}",
            status="failure",
            resource_type="cmdb_server",
            resource_id=str(server_id),
            detail={
                "reason": "provider_auth_error",
                "provider_message": redact_provider_error(exc, max_len=300),
                "integration_id": integration_id_str,
                "instance_id": instance_id,
            },
        )
        await db.commit()
        raise AppException(
            code="CLOUD_LIFECYCLE_AUTH_ERROR",
            message=(
                f"云服务商凭证拒绝该操作:{redact_provider_error(exc, max_len=200)}"
            ),
            status_code=502,
        ) from exc
    except ValueError as exc:
        # Driver-side guard: instance row has no region/project — surface
        # as a 422 so the admin sees what's missing rather than a 500.
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=user.id,
            action=f"cmdb.server.{action}",
            status="failure",
            resource_type="cmdb_server",
            resource_id=str(server_id),
            detail={
                "reason": "missing_required_field",
                "provider_message": str(exc)[:300],
            },
        )
        await db.commit()
        raise ValidationError(str(exc)) from exc
    except ProviderTransientError as exc:
        redacted = redact_provider_error(exc, max_len=200)
        logger.error(
            "cmdb.server.%s provider transient failure: %s", action, redacted,
        )
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=user.id,
            action=f"cmdb.server.{action}",
            status="failure",
            resource_type="cmdb_server",
            resource_id=str(server_id),
            detail={
                "reason": "provider_transient_error",
                "provider_message": redacted,
                "integration_id": integration_id_str,
                "instance_id": instance_id,
            },
        )
        await db.commit()
        raise AppException(
            code="CLOUD_LIFECYCLE_FAILED",
            message=(
                f"云服务商拒绝该操作:{redacted}"
                if redacted
                else "Cloud provider rejected the request (no detail returned)."
            ),
            status_code=502,
        ) from exc
    except Exception as exc:
        # Catch-all for unexpected driver bugs (e.g. an SDK raising a
        # non-Provider* exception). Redact before logging — the bare
        # exception args may carry signed request URLs / response
        # bodies through to ELK.
        redacted = redact_provider_error(exc, max_len=200)
        logger.error(
            "cmdb.server.%s unexpected failure (%s): %s",
            action, type(exc).__name__, redacted,
        )
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=user.id,
            action=f"cmdb.server.{action}",
            status="failure",
            resource_type="cmdb_server",
            resource_id=str(server_id),
            detail={
                "reason": "unexpected_error",
                "error_class": type(exc).__name__,
                "provider_message": redacted,
                "integration_id": integration_id_str,
                "instance_id": instance_id,
            },
        )
        await db.commit()
        raise AppException(
            code="CLOUD_LIFECYCLE_FAILED",
            message=(
                f"云服务商拒绝该操作:{redacted}"
                if redacted
                else "Cloud provider rejected the request (no detail returned)."
            ),
            status_code=502,
        ) from exc
    finally:
        await provider.close()

    # Optimistic transitional status — the next sync cron reconciles
    # against the cloud's authoritative state. For destroy we ALSO
    # soft-delete the row so it drops from the active list immediately;
    # the recycle bin still holds it for restore-on-mistake.
    new_status = _LIFECYCLE_TRANSITIONAL_STATUS[action]
    server.status = new_status
    server.last_modified_by_person_id = str(user.id)
    server.last_modified_at = datetime.now(UTC)
    if action == "destroy":
        server.deleted_at = datetime.now(UTC)

    # Single commit covers the state change AND the success audit row.
    # Earlier shape committed the state first then logged the audit
    # row — the audit write stayed in the session until the next
    # commit, which never came. The audit log was lost on every
    # success path; the failure paths above were already broken the
    # same way. The single-commit shape used here is the same pattern
    # the DNS enable/disable + force-sweep endpoints use.
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action=f"cmdb.server.{action}",
        status="success",
        resource_type="cmdb_server",
        resource_id=str(server_id),
        detail={
            "integration_id": integration_id_str,
            "instance_id": instance_id,
            "region": region,
            "project_id": project_id,
            "new_status": new_status,
            "soft_deleted": action == "destroy",
        },
    )
    await db.flush()
    await db.commit()
    await db.refresh(server)

    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [server])
    integ_meta = await _resolve_integration_meta(
        db, [str(server.integration_id)],
    )
    return _attach_integration_meta(
        _augment_response(
            ServerResponse.model_validate(server).model_dump(), server, labels, owner_meta,
        ),
        server, integ_meta,
    )


# ``require_browser_session`` gates these four routes — same step-up
# auth applied to the other catastrophic-if-stolen admin actions
# (force-sweep, role-mutation, PII reveal). A leaked cloud_admin PAT
# could otherwise nuke production infrastructure via a bare ``curl``;
# requiring a real Google-SSO browser session blocks that escalation
# path. MCP callers are still able to invoke the matching
# ``server_*`` tools — their PAT step-up happened at mint time and
# the runtime applies its own authz layer.


@router.post("/servers/{server_id}/power-on")
async def server_power_on(
    server_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _csrf: None = Depends(require_browser_session),
):
    item = await _execute_server_lifecycle(
        "power_on", server_id, request, user, db,
    )
    return success_response(data=item, request=request)


@router.post("/servers/{server_id}/power-off")
async def server_power_off(
    server_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _csrf: None = Depends(require_browser_session),
):
    item = await _execute_server_lifecycle(
        "power_off", server_id, request, user, db,
    )
    return success_response(data=item, request=request)


@router.post("/servers/{server_id}/restart")
async def server_restart(
    server_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _csrf: None = Depends(require_browser_session),
):
    item = await _execute_server_lifecycle(
        "restart", server_id, request, user, db,
    )
    return success_response(data=item, request=request)


@router.post("/servers/{server_id}/destroy")
async def server_destroy(
    server_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _csrf: None = Depends(require_browser_session),
):
    """Terminate the cloud instance AND soft-delete the cmdb row.

    Irreversible on the cloud side (TerminateUHostInstance /
    DeleteInstance / TerminateInstances). The cmdb row is recoverable
    via the recycle bin (``/cmdb/recycle-bin/server``) but the
    underlying VM is gone.
    """
    item = await _execute_server_lifecycle(
        "destroy", server_id, request, user, db,
    )
    return success_response(data=item, request=request)


# ---------------------------------------------------------------------------
# Domains
# ---------------------------------------------------------------------------


@router.get("/domains")
async def list_domains(
    request: Request,
    integration_id: str | None = None,
    expiring_within_days: int | None = None,
    owner_person_id: str | None = None,
    search: str | None = None,
    page: int = 1,
    # le=500 (not 200): MyCloudDomainsPage's expiring-soon banner
    # fetches up to 500 rows in a single call so we can correctly
    # count owners with > 200 domains in the window (see PR #353).
    # The fleet domain table is the only cloud-resource list where
    # plain employees see ALL rows (no auto-demote — see RBAC note
    # below). With ~60 prod domains today the worst case is the
    # fleet pulling at most 60 rows; the limit is kept high to leave
    # headroom for future growth and to match the admin path.
    page_size: int = Query(50, ge=1, le=500),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    # RBAC for /cmdb/domains diverges from the sibling list endpoints
    # (servers / DNS / CDN / LB / IAM / object storage) which all
    # auto-demote non-admins to "see only what you own or what your
    # subordinates own". Domains are intentionally the exception:
    #   * The domain_name itself is company-public knowledge —
    #     ``pingcap.com`` / ``tidb-community.com`` / ``pingkai.cn``
    #     are printed on business cards, in marketing material, on
    #     every product page. They carry zero reconnaissance value
    #     beyond what a ``dig`` already exposes.
    #   * The other columns surfaced by ``DomainResponse`` (registrar
    #     name, expiry date, auto-renew flag, ICP filing number,
    #     environment label) are operational facts every employee
    #     needs to see — the use-case driving this change was a
    #     non-admin employee filing a ticket "is pingkai.cn about to
    #     expire? auto-renew on?" The prior owner-demote silently
    #     hid every row the employee didn't personally own,
    #     including domains that show ``status='inactive'`` and
    #     ``expiry_date`` in the past — exactly the rows IT needs
    #     the broadest eyeballs on.
    # Sibling endpoints KEEP the owner demote because they leak:
    #   * Hostnames + private IPs (servers, DNS A/AAAA)
    #   * DKIM/SPF TXT contents (DNS records)
    #   * Internal load-balancer endpoints + listener ports
    #   * IAM ARNs + attached-policy fingerprints
    #   * Bucket ACL state
    # — all of which ARE reconnaissance bait. Domain != DNS record:
    # the rule split here is deliberate.
    # See decision_domain_public_visibility.md.
    # The caller-supplied ``owner_person_id`` filter is honoured for
    # everyone (admins use it to audit one owner; employees use it
    # to list "my domains" via owner_person_id='me' on the MCP path,
    # which the MCP tool maps to str(user.id) before reaching here).
    domains, total = await CmdbService.list_domains(
        db,
        integration_id=integration_id,
        expiring_within_days=expiring_within_days,
        owner_person_id=owner_person_id,
        owner_person_ids=None,
        search=search,
        page=page,
        page_size=page_size,
    )
    # Resolve owner / last-modified labels so the table renders 张三 instead
    # of a bare UUID. Mirrors /servers, /load-balancers, /cdn-domains —
    # this endpoint was the lone hold-out which is why CloudDomainsPage's
    # 负责人 column displayed "-" and the edit drawer's UserSelect showed
    # the raw owner_person_id until a search call repopulated the option.
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, domains)
    data = [
        _augment_response(DomainResponse.model_validate(d).model_dump(), d, labels, owner_meta)
        for d in domains
    ]
    return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request)


@router.get("/domains/export")
async def export_domains(
    request: Request,
    integration_id: str | None = None,
    expiring_within_days: int | None = None,
    search: str | None = None,
    owner_person_id: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Export all matching domains as CSV (up to 5000 rows).

    Visibility mirrors ``list_domains`` — every authenticated employee
    with ``can_view_cloud_resources`` can export the same set they see
    in the table. No owner demote (domains are company-public; see
    ``list_domains`` for the RBAC rationale).
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    domains, _ = await CmdbService.list_domains(
        db,
        integration_id=integration_id,
        expiring_within_days=expiring_within_days,
        owner_person_id=owner_person_id,
        owner_person_ids=None,
        search=search,
        page=1,
        page_size=5000,
    )
    owner_labels = await _resolve_owner_labels(
        db, [d.owner_person_id for d in domains if d.owner_person_id],
    )
    rows = []
    for d in domains:
        rows.append({
            "domain_name": d.domain_name,
            "registrar": d.registrar or "",
            "expiry_date": str(d.expiry_date) if d.expiry_date else "",
            "auto_renew": str(d.auto_renew),
            "icp_filing_no": d.icp_filing_no or "",
            "status": d.status,
            "owner": owner_labels.get(d.owner_person_id, "") if d.owner_person_id else "",
            "last_synced_at": str(d.last_synced_at)[:19],
        })
    return _csv_response(rows, "cmdb_domains.csv")


@router.get("/domains/selectable-for-dns")
async def list_selectable_dns_zones(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Return company-registered zones an employee can target for DNS.

    Used by the self-service DNS 申请 modal Zone picker. Each zone is
    accompanied by its resolved authoritative-DNS provider — NOT the
    registrar — so the submit flow routes the eventual write to the
    correct cloud tenant.

    Filtering:

    * ``cmdb_domains.allow_dns_request=True`` — admin gate for which
      zones are open to self-service. Defaults True; admins can lock
      down prod zones explicitly.
    * The integration that **registered** the domain in ``cmdb_domains``
      is the registrar metadata. The integration that **hosts DNS
      records** for the zone is resolved via
      ``CmdbService.resolve_dns_integration_for_zone``. The two can
      legitimately differ (UCloud-registered domain hosted on Route53,
      DNSPod-hosted zone whose registrar is Tencent 域名注册).
    * Zones where the registrar's tenant is disabled are still
      eligible if the DNS-provider tenant is enabled — the registrar
      tenant's state is orthogonal to DNS-write capability.
    * Zones with no resolved DNS provider are returned with
      ``integration_id=None``; the FE filters them out of the
      employee picker (employees can't seed the first record
      cold; admin direct-create is the path for that).

    Prod incident 2026-06-01: the prior implementation returned
    ``cmdb_domains.integration_id`` as ``integration_id``, which is
    the registrar (e.g. UCloud for tikv.io). Employee submits
    flowed the registrar id into ``CmdbDnsRecordRequest`` and the
    eventual execute called the wrong cloud SDK. This corrected
    response unifies admin + self-service on the same DNS-provider
    resolution.
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbDomain
    # Pull every admin-allowed zone (registrar tenant filter dropped
    # intentionally — see docstring; DNS provider resolution does the
    # right scoping).
    result = await db.execute(
        select(CmdbDomain.domain_name).where(
            CmdbDomain.deleted_at.is_(None),
            CmdbDomain.allow_dns_request.is_(True),
        ).distinct().order_by(CmdbDomain.domain_name)
    )
    zone_names = [r[0] for r in result.all()]
    if not zone_names:
        return success_response(data=[], request=request)

    # Resolve the DNS provider per zone in one pass: load every
    # (zone, integration) pair from cmdb_dns_records, pick the
    # winner by record count. Avoids a per-zone round-trip.
    from sqlalchemy import func as sa_func

    from app.domains.cmdb.models import CmdbDnsRecord
    active_ids = await _active_cloud_integration_ids(db)
    if not active_ids:
        # No enabled cloud tenants → every zone resolves to None.
        # Skip the records query entirely so we don't waste a round-trip
        # AND mypy sees the records loop branch as having a typed
        # ColumnElement where-clause.
        records_rows: list[Any] = []
    else:
        records_rows = list((
            await db.execute(
                select(
                    CmdbDnsRecord.zone_name,
                    CmdbDnsRecord.integration_id,
                    sa_func.count(CmdbDnsRecord.id).label("rc"),
                ).where(
                    CmdbDnsRecord.deleted_at.is_(None),
                    CmdbDnsRecord.zone_name.in_(zone_names),
                    # Only resolve to enabled tenants — a disabled tenant
                    # with a stale historical record must not silently
                    # become the routing target.
                    CmdbDnsRecord.integration_id.in_(active_ids),
                ).group_by(CmdbDnsRecord.zone_name, CmdbDnsRecord.integration_id),
            )
        ).all())
    zone_to_winner: dict[str, tuple[str, int]] = {}
    for zone, integ, rc in records_rows:
        current = zone_to_winner.get(zone)
        if current is None or rc > current[1]:
            zone_to_winner[zone] = (str(integ), int(rc))

    integ_ids = {w[0] for w in zone_to_winner.values()}
    integ_meta: dict[str, tuple[str, str]] = {}
    if integ_ids:
        rows = (
            await db.execute(
                select(
                    Integration.id, Integration.name, Integration.provider,
                ).where(Integration.id.in_(integ_ids)),
            )
        ).all()
        integ_meta = {str(r[0]): (r[1], r[2]) for r in rows}

    data = []
    for zone in zone_names:
        winner = zone_to_winner.get(zone)
        if winner is None:
            data.append(DomainSelectableForDns(
                domain_name=zone,
                integration_id=None,
                integration_name=None,
                provider=None,
            ).model_dump())
            continue
        integ_id, _rc = winner
        name, provider = integ_meta.get(integ_id, ("", ""))
        data.append(DomainSelectableForDns(
            domain_name=zone,
            integration_id=integ_id,
            integration_name=name or None,
            provider=provider or None,
        ).model_dump())
    return success_response(data=data, request=request)


@router.get("/domains/{domain_id}")
async def get_domain(
    domain_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Visibility mirrors ``list_domains`` — every authenticated employee
    # with ``can_view_cloud_resources`` can fetch any domain by id.
    # The prior per-row owner check was over-correction copied from
    # ``get_server`` / ``get_dns_record`` (which DO leak internal
    # hostnames / IPs / DKIM TXT). Domains are public identifiers;
    # see the ``list_domains`` docstring for the full rationale.
    # Tenant scoping is enforced inside the disabled-tenant filter
    # used by list_domains (``_active_cloud_integration_ids``) — the
    # by-id path does NOT apply that filter because admins
    # legitimately audit one frozen tenant's rows by id (matches the
    # 2026-05-25 prod incident fix for /servers/{id}).
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbDomain
    result = await db.execute(
        select(CmdbDomain).where(
            CmdbDomain.id == domain_id,
            CmdbDomain.deleted_at.is_(None),
        )
    )
    domain = result.scalar_one_or_none()
    if not domain:
        raise NotFoundError("Domain", str(domain_id))
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [domain])
    item = _augment_response(
        DomainResponse.model_validate(domain).model_dump(), domain, labels, owner_meta,
    )
    return success_response(data=item, request=request)


@router.patch("/domains/{domain_id}")
async def annotate_domain(
    domain_id: uuid.UUID,
    body: DomainAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    # Capture before_state BEFORE the mutation so the audit row
    # records who owned the domain previously. Without this, the
    # audit_log only carries the patch payload, and reconstructing
    # "who owned example.com last month" requires walking
    # ``cmdb_network_owner_history`` row by row.
    from app.domains.cmdb.models import CmdbDomain as _CmdbDomain  # noqa: PLC0415
    pre = await db.scalar(
        select(_CmdbDomain).where(
            _CmdbDomain.id == domain_id,
            _CmdbDomain.deleted_at.is_(None),
        ),
    )
    before_owner = pre.owner_person_id if pre else None
    before_icp = pre.icp_filing_no if pre else None
    before_env = pre.environment if pre else None
    before_allow = pre.allow_dns_request if pre else None
    # Migration 218: capture the sync-owned columns + provenance flag
    # in the before-snapshot so a future forensic query ("who flipped
    # pingcap.com auto-renew off on date X?") can answer from the
    # audit log alone, without walking ``cmdb_network_owner_history``
    # (which only tracks owner). ``data_source`` is included because
    # touching any of the 3 sync-owned fields flips it to ``manual``
    # — capturing the transition makes a "this row went stale because
    # an admin overrode" trail visible to auditors.
    before_registrar = pre.registrar if pre else None
    before_expiry = pre.expiry_date.isoformat() if pre and pre.expiry_date else None
    before_auto_renew = pre.auto_renew if pre else None
    before_data_source = pre.data_source if pre else None
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    domain = await CmdbService.annotate_domain(
        db, str(domain_id), body, actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    if not domain:
        raise NotFoundError("Domain", str(domain_id))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.domain.annotate",
        status="success",
        resource_type="cmdb_domain",
        resource_id=str(domain_id),
        detail={
            "patch": body.model_dump(exclude_unset=True, mode="json"),
            "before": {
                "owner_person_id": before_owner,
                "icp_filing_no": before_icp,
                "environment": before_env,
                "allow_dns_request": before_allow,
                "registrar": before_registrar,
                "expiry_date": before_expiry,
                "auto_renew": before_auto_renew,
                "data_source": before_data_source,
            },
            "after": {
                "owner_person_id": domain.owner_person_id,
                "icp_filing_no": domain.icp_filing_no,
                "environment": domain.environment,
                "allow_dns_request": domain.allow_dns_request,
                "registrar": domain.registrar,
                "expiry_date": domain.expiry_date.isoformat() if domain.expiry_date else None,
                "auto_renew": domain.auto_renew,
                "data_source": domain.data_source,
            },
        },
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [domain])
    item = _augment_response(
        DomainResponse.model_validate(domain).model_dump(), domain, labels, owner_meta,
    )
    return success_response(data=item, request=request)


# ---------------------------------------------------------------------------
# TLS/SSL Certificates
# ---------------------------------------------------------------------------


def _cert_item(row, labels, owner_meta=None, resolution=None) -> dict:
    from app.domains.cmdb.cert_service import (  # noqa: PLC0415 — sibling
        effective_cert_status,
    )

    item = _augment_response(
        CertificateResponse.model_validate(row).model_dump(mode="json"),
        row, labels, owner_meta,
    )
    # Same "the stored snapshot is not the truth" treatment the owner chain
    # gets below: the lifecycle status is derived from ``not_after`` on read,
    # because the stored column is only written when a probe runs.
    effective_status = effective_cert_status(row)
    item["status_stale"] = effective_status != row.status
    item["status"] = effective_status
    if resolution is not None:
        # Ownership is live-resolved (DNS record → zone domain → network
        # owner), not read from the stored column. Overwrite the snapshot
        # fields with the resolved chain + provenance so the UI shows the
        # current responsible person and where they came from.
        eff = resolution.owner_person_id
        item["owner_person_id"] = eff
        item["owner_source"] = resolution.owner_source
        item["owner_is_manual"] = resolution.owner_is_manual
        item["matched_zone"] = resolution.matched_zone
        item["owner_label"] = labels.get(str(eff)) if eff else None
        if owner_meta is not None and eff:
            m = owner_meta.get(str(eff))
            if m:
                item["owner_employee_no"] = m["employee_no"]
                item["owner_employment_status"] = m["employment_status"]
        else:
            item["owner_employee_no"] = None
            item["owner_employment_status"] = None
        dom_owner = resolution.domain_owner_person_id
        item["domain_owner_person_id"] = dom_owner
        item["domain_owner_label"] = labels.get(str(dom_owner)) if dom_owner else None
    return item


async def _resolve_cert_responses(db: AsyncSession, rows, resolutions) -> list[dict]:
    """Build certificate response dicts with live-resolved ownership.

    Resolves person labels + leaver-meta for the union of effective owners
    and domain owners across ``resolutions`` (one batch query each), then
    renders each row.
    """
    pids: list[str] = []
    for r in rows:
        res = resolutions.get(str(r.id))
        if res is None:
            continue
        if res.owner_person_id:
            pids.append(str(res.owner_person_id))
        if res.domain_owner_person_id:
            pids.append(str(res.domain_owner_person_id))
        if getattr(r, "last_modified_by_person_id", None):
            pids.append(str(r.last_modified_by_person_id))
    labels = await _resolve_owner_labels(db, pids)
    owner_meta = await _resolve_owner_meta(db, pids)
    # Batch-load expiry-alert snooze state so the UI can badge "续期中".
    ack_until: dict[str, str] = {}
    if rows:
        from app.domains.cmdb.models import (
            EXPIRY_RESOURCE_CERTIFICATE,
            CmdbExpiryAlertState,
        )
        ack_rows = (await db.execute(
            select(CmdbExpiryAlertState).where(
                CmdbExpiryAlertState.resource_type == EXPIRY_RESOURCE_CERTIFICATE,
                CmdbExpiryAlertState.resource_id.in_([str(r.id) for r in rows]),
            )
        )).scalars().all()
        for s in ack_rows:
            if s.acknowledged_until is not None:
                ack_until[s.resource_id] = s.acknowledged_until.isoformat()
    out = []
    for r in rows:
        item = _cert_item(r, labels, owner_meta, resolutions.get(str(r.id)))
        item["acknowledged_until"] = ack_until.get(str(r.id))
        out.append(item)
    return out


@router.get("/certificates")
async def list_certificates(
    request: Request,
    status: str | None = None,
    source: str | None = None,
    owner_person_id: str | None = None,
    environment: str | None = None,
    search: str | None = None,
    page: int = 1,
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    # Auto-demote: certs carry probe hostnames + SAN lists (recon bait,
    # like DNS records / servers), so non-privileged callers only see
    # certs they own or their subordinates own. Mirrors /dns-records.
    owner_ids_scope: list[str] | None = None
    if not can_read_cloud(user.role_names):
        subs = await get_subordinate_person_ids(db, str(user.id))
        owner_ids_scope = [str(user.id), *subs]
        owner_person_id = None
    rows, total, resolutions = await cert_service.list_certificates(
        db,
        status=status,
        source=source,
        owner_person_id=owner_person_id,
        environment=environment,
        search=search,
        owner_ids_scope=owner_ids_scope,
        page=page,
        page_size=page_size,
    )
    data = await _resolve_cert_responses(db, rows, resolutions)
    return paginated_response(
        data=data, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/certificates/stats")
async def certificate_stats(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """KPI counts for the admin certificate dashboard (admins/auditors)."""
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from datetime import UTC as _UTC
    from datetime import datetime as _dt
    from datetime import timedelta as _td

    from sqlalchemy import func

    from app.domains.cmdb.models import (
        CERT_STATUS_UNREACHABLE,
        CmdbCertificate,
    )

    now = _dt.now(_UTC)
    base = select(func.count()).select_from(CmdbCertificate).where(
        CmdbCertificate.deleted_at.is_(None),
    )
    total = int((await db.execute(base)).scalar_one())
    expired = int((await db.execute(
        base.where(CmdbCertificate.not_after.is_not(None),
                   CmdbCertificate.not_after < now)
    )).scalar_one())
    within7 = int((await db.execute(
        base.where(CmdbCertificate.not_after >= now,
                   CmdbCertificate.not_after < now + _td(days=7))
    )).scalar_one())
    within30 = int((await db.execute(
        base.where(CmdbCertificate.not_after >= now,
                   CmdbCertificate.not_after < now + _td(days=30))
    )).scalar_one())
    unreachable = int((await db.execute(
        base.where(CmdbCertificate.status == CERT_STATUS_UNREACHABLE)
    )).scalar_one())

    # Owner coverage — governance signal: how many certs have NO
    # resolvable responsible person (no DNS/domain owner, no override, no
    # network-owner fallback configured). Resolved live over all rows.
    from app.domains.cmdb.cert_owner_resolver import (
        OWNER_SOURCE_NONE,
        CertOwnerResolver,
    )
    from app.domains.cmdb.network_notifications import resolve_network_owner_id

    all_rows = list((await db.execute(
        select(CmdbCertificate).where(CmdbCertificate.deleted_at.is_(None))
    )).scalars().all())
    network_owner_id = await resolve_network_owner_id(db)
    resolver = await CertOwnerResolver.build(db, network_owner_id=network_owner_id)
    ownerless = sum(
        1 for r in all_rows
        if resolver.resolve(r).owner_source == OWNER_SOURCE_NONE
    )
    return success_response(
        data={
            "total": total,
            "expired": expired,
            "expiring_within_7d": within7,
            "expiring_within_30d": within30,
            "unreachable": unreachable,
            "ownerless": ownerless,
        },
        request=request,
    )


async def _resolve_one_cert(db, row):
    """Resolve a single certificate's live ownership chain."""
    from app.domains.cmdb.cert_owner_resolver import CertOwnerResolver
    from app.domains.cmdb.network_notifications import resolve_network_owner_id

    network_owner_id = await resolve_network_owner_id(db)
    resolver = await CertOwnerResolver.build(db, network_owner_id=network_owner_id)
    return resolver.resolve(row)


async def _load_cert_for_caller(db, cert_id, user):
    """Fetch a cert + enforce scope. Non-privileged callers may only read
    certs whose LIVE-RESOLVED owner is themselves or a subordinate."""
    row = await cert_service.get_certificate(db, str(cert_id))
    if row is None:
        raise NotFoundError("Certificate", str(cert_id))
    resolution = await _resolve_one_cert(db, row)
    if not can_read_cloud(user.role_names):
        subs = set(await get_subordinate_person_ids(db, str(user.id)))
        allowed = {str(user.id), *subs}
        if str(resolution.owner_person_id or "") not in allowed:
            raise ForbiddenError()
    return row, resolution


@router.get("/certificates/{cert_id}")
async def get_certificate(
    cert_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    row, resolution = await _load_cert_for_caller(db, cert_id, user)
    data = await _resolve_cert_responses(db, [row], {str(row.id): resolution})
    return success_response(data=data[0], request=request)


@router.post("/certificates", status_code=201)
async def create_certificate(
    body: CertificateCreate,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    row = await cert_service.create_manual_certificate(
        db,
        common_name=body.common_name,
        not_after=body.not_after,
        not_before=body.not_before,
        issuer=body.issuer,
        sans=body.sans,
        owner_person_id=str(body.owner_person_id) if body.owner_person_id else None,
        environment=body.environment,
        remark=body.remark,
        actor_person_id=str(user.id),
    )
    await db.flush()
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.certificate.create",
        status="success",
        resource_type="cmdb_certificate",
        resource_id=str(row.id),
        detail={"create": body.model_dump(exclude_unset=True, mode="json")},
    )
    await db.commit()
    resolution = await _resolve_one_cert(db, row)
    data = await _resolve_cert_responses(db, [row], {str(row.id): resolution})
    return success_response(data=data[0], request=request)


@router.patch("/certificates/{cert_id}")
async def annotate_certificate(
    cert_id: uuid.UUID,
    body: CertificateAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    row = await cert_service.get_certificate(db, str(cert_id))
    if row is None:
        raise NotFoundError("Certificate", str(cert_id))
    before = {
        "owner_person_id": row.owner_person_id,
        "environment": row.environment,
        "remark": row.remark,
    }
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    await cert_service.annotate_certificate(
        db, row,
        fields_set=set(body.model_fields_set),
        owner_person_id=str(body.owner_person_id) if body.owner_person_id else None,
        environment=body.environment,
        remark=body.remark,
        actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.certificate.annotate",
        status="success",
        resource_type="cmdb_certificate",
        resource_id=str(cert_id),
        detail={
            "patch": body.model_dump(exclude_unset=True, mode="json"),
            "before": before,
            "after": {
                "owner_person_id": row.owner_person_id,
                "environment": row.environment,
                "remark": row.remark,
            },
        },
    )
    await db.commit()
    resolution = await _resolve_one_cert(db, row)
    data = await _resolve_cert_responses(db, [row], {str(row.id): resolution})
    return success_response(data=data[0], request=request)


@router.delete("/certificates/{cert_id}")
async def delete_certificate(
    cert_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Soft-delete a certificate row. A probed cert that is still served
    will reappear on the next sweep — use this to retire manual rows or
    decommissioned endpoints."""
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from datetime import UTC as _UTC
    from datetime import datetime as _dt

    row = await cert_service.get_certificate(db, str(cert_id))
    if row is None:
        raise NotFoundError("Certificate", str(cert_id))
    row.deleted_at = _dt.now(_UTC)
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.certificate.delete",
        status="success",
        resource_type="cmdb_certificate",
        resource_id=str(cert_id),
        detail={"source": row.source, "common_name": row.common_name},
    )
    await db.commit()
    return success_response(data={"id": str(cert_id), "deleted": True}, request=request)


# ---------------------------------------------------------------------------
# Expiry-alert acknowledge / snooze (cert + domain)
# ---------------------------------------------------------------------------

_EXPIRY_RESOURCE_TYPES = {"certificate", "domain"}


@router.post("/expiry/{resource_type}/{resource_id}/acknowledge")
async def acknowledge_expiry_alert(
    resource_type: str,
    resource_id: str,
    request: Request,
    days: int = Query(14, ge=1, le=90),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Mark a cert / domain "renewing" — snooze its expiry alerts (incl.
    escalation) for ``days`` days. Allowed for cloud admins or for one of
    the resource's responsible people (本人 / 域名负责人 / lead)."""
    if resource_type not in _EXPIRY_RESOURCE_TYPES:
        raise NotFoundError("ExpiryResource", resource_type)
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.expiry_alert_engine import (
        acknowledge_expiry,
        resolve_expiry_owners,
    )

    if not can_manage_cloud(user.role_names):
        owners = await resolve_expiry_owners(db, resource_type, resource_id)
        if str(user.id) not in owners:
            raise ForbiddenError()
    row = await acknowledge_expiry(
        db, resource_type=resource_type, resource_id=resource_id,
        actor_person_id=str(user.id), days=days,
    )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    await AuditService(db).log(
        trace_id=trace_id, actor_type="user", actor_user_id=user.id,
        action="cmdb.expiry.acknowledge", status="success",
        resource_type=f"cmdb_{resource_type}", resource_id=resource_id,
        detail={"acknowledged_until": row.acknowledged_until.isoformat(), "days": days},
    )
    await db.commit()
    return success_response(
        data={"acknowledged_until": row.acknowledged_until.isoformat()},
        request=request,
    )


@router.post("/expiry/ack")
async def acknowledge_expiry_via_token(
    request: Request,
    token: str = Query(...),
    db: AsyncSession = Depends(get_db),
):
    """One-click no-login acknowledge from the Feishu card magic link. The
    signed action token (purpose=``expiry_ack``) carries the acting person
    + the ``{resource_type}:{resource_id}`` it authorises."""
    from app.core.dependencies import enforce_action_token_subject
    from app.core.security import decode_action_token
    from app.domains.cmdb.expiry_alert_engine import acknowledge_expiry

    try:
        payload = decode_action_token(token, expected_purpose="expiry_ack")
        # Departed / force-logged-out subject's stale link must not still ack.
        await enforce_action_token_subject(db, payload)
    except Exception as exc:  # noqa: BLE001 — bad/expired/departed → 403
        raise ForbiddenError("Invalid or expired acknowledge link.") from exc
    res = str(payload.get("res") or "")
    rtype, _, rid = res.partition(":")
    if rtype not in _EXPIRY_RESOURCE_TYPES or not rid:
        raise NotFoundError("ExpiryResource", res)
    # Resource ids are uuids — reject a malformed/oddly-shaped token target
    # rather than letting it fall through to a confusing not-found later.
    try:
        actor_uuid = uuid.UUID(str(payload["sub"]))
        uuid.UUID(rid)
    except (ValueError, KeyError, TypeError) as exc:
        raise NotFoundError("ExpiryResource", res) from exc
    row = await acknowledge_expiry(
        db, resource_type=rtype, resource_id=rid,
        actor_person_id=str(actor_uuid), days=14,
    )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    await AuditService(db).log(
        trace_id=trace_id, actor_type="user",
        actor_user_id=actor_uuid,
        action="cmdb.expiry.acknowledge", status="success",
        resource_type=f"cmdb_{rtype}", resource_id=rid,
        detail={"via": "action_token",
                "acknowledged_until": row.acknowledged_until.isoformat()},
    )
    await db.commit()
    return success_response(
        data={"acknowledged_until": row.acknowledged_until.isoformat(),
              "resource_type": rtype},
        request=request,
    )


# ---------------------------------------------------------------------------
# DNS Records
# ---------------------------------------------------------------------------


@router.get("/dns-zones")
async def list_dns_zones(
    request: Request,
    integration_id: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # DNS zones are tenant-internal naming; admins/auditors only.
    # Employees see their own records via /dns-records (auto-demoted)
    # and the per-record-owner /me path; the zone catalog itself is
    # not in their threat-relevant scope.
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    zones = await CmdbService.list_dns_zones(db, integration_id=integration_id)
    return success_response(data=zones, request=request)


@router.get("/dns-records")
async def list_dns_records(
    request: Request,
    integration_id: str | None = None,
    zone_name: str | None = None,
    record_type: str | None = None,
    search: str | None = None,
    owner_person_id: str | None = None,
    page: int = 1,
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    # DNS records are a company-public asset (ADR 2026-06-09): every caller
    # who passes the cloud-view gate sees the FULL list, regardless of owner.
    # Plain employees get read-only visibility — write / disable / delete
    # still require can_manage_cloud, and the my-cloud-dns page only renders
    # row actions on the caller's own records. Unlike /servers and /domains
    # (which stay owner-scoped), DNS is intentionally not recon-protected.
    # ``owner_person_id`` remains an OPTIONAL view filter (admin "filter by
    # owner" UserSelect); it is NOT a security boundary.
    records, total = await CmdbService.list_dns_records(
        db,
        integration_id=integration_id,
        zone_name=zone_name,
        record_type=record_type,
        search=search,
        owner_person_id=owner_person_id,
        page=page,
        page_size=page_size,
    )
    # Populate owner_label so the my-cloud-dns page can render
    # "owner: 张三" without a per-row N+1. Same pattern as
    # /dns-records/{id} below and the servers / domains list endpoints.
    owner_labels = await _resolve_owner_labels(
        db, [r.owner_person_id for r in records if r.owner_person_id],
    )
    owner_meta = await _resolve_owner_meta(
        db, [r.owner_person_id for r in records if r.owner_person_id],
    )
    data = []
    for r in records:
        item = DnsRecordResponse.model_validate(r).model_dump()
        if r.owner_person_id:
            item["owner_label"] = owner_labels.get(r.owner_person_id)
            m = owner_meta.get(r.owner_person_id)
            if m:
                item["owner_employee_no"] = m["employee_no"]
                item["owner_employment_status"] = m["employment_status"]
        data.append(item)
    return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request)


@router.get("/dns-records/{record_id}")
async def get_dns_record(
    record_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbDnsRecord
    result = await db.execute(
        select(CmdbDnsRecord).where(
            CmdbDnsRecord.id == record_id,
            CmdbDnsRecord.deleted_at.is_(None),
        )
    )
    record = result.scalar_one_or_none()
    if not record:
        raise NotFoundError("DnsRecord", str(record_id))
    # DNS records are a company-public asset (ADR 2026-06-09): any caller
    # who passes the cloud-view gate may read any record's detail, exactly
    # like list_dns_records returns the full list. Write / disable / delete
    # still require can_manage_cloud; detail read is intentionally not
    # owner-scoped (a per-row owner check here contradicted the list and
    # 404'd employees on records they could see in the list).
    item = DnsRecordResponse.model_validate(record).model_dump()
    if record.owner_person_id:
        labels = await _resolve_owner_labels(db, [record.owner_person_id])
        item["owner_label"] = labels.get(record.owner_person_id)
        owner_meta = await _resolve_owner_meta(db, [record.owner_person_id])
        m = owner_meta.get(str(record.owner_person_id))
        if m:
            item["owner_employee_no"] = m["employee_no"]
            item["owner_employment_status"] = m["employment_status"]
    return success_response(data=item, request=request)


async def _load_provider_for_dns(db: AsyncSession, integration_id: str):
    """Load a live provider instance for DNS write operations.

    Credentials live on the ``integrations`` row now; decrypt them and
    feed the concrete :class:`CloudProvider` just like the sync path does.
    """
    integration = await _load_cloud_integration(db, integration_id)
    # Fail fast if the integration doesn't advertise the dns_records
    # capability. Resolution is the same as the sync scheduler's —
    # ``resolve_sync_capabilities`` falls back to the driver class's
    # ``supported_capabilities()`` when the admin-set column is empty.
    # Without this check, a misdirected ``POST /dns-records`` to a
    # cloud account that doesn't host DNS (e.g. Azure-only tenant)
    # would proceed to instantiate the driver and only fail at the
    # final ``provider.create_dns_record(...)`` call with an
    # opaque ``CapabilityNotSupported``.
    from app.domains.cmdb.scheduled_jobs_provisioning import (
        resolve_sync_capabilities,
    )
    if "dns_records" not in resolve_sync_capabilities(integration):
        # Truncate ``integration.name`` defensively: admin-set free
        # text could contain noisy or sensitive material (some teams
        # paste creds into the name field). The error toast surfaces
        # to non-admins on the self-service path; ``provider`` (enum
        # slug) + truncated name + opaque id is enough to act on
        # without making the error a credential disclosure vector.
        safe_name = (integration.name or "")[:80]
        raise AppException(
            code="CLOUD_INTEGRATION_LACKS_DNS_CAPABILITY",
            message=(
                f"Integration '{safe_name}' ({integration.provider}) "
                "does not advertise the dns_records capability. Either pick "
                "a different cloud tenant or have the IT Admin enable "
                "dns_records on this integration."
            ),
            status_code=400,
        )
    # Read-only archival tenants (ksyun, historical ucloud before any
    # creds are attached) have no stored secrets — instantiate the
    # provider stub directly so DNS writes surface a capability error
    # rather than a generic "integration missing credentials" 404.
    archival_providers = {"ksyun", "ucloud"}
    if integration.provider in archival_providers and not integration.secrets_encrypted:
        creds = ProviderCredentials(
            provider_id=str(integration.id),
            provider_type=integration.provider,
            account_alias=integration.name,
            fields={},
        )
        return get_provider_class(integration.provider)(creds)
    if not integration.secrets_encrypted:
        raise AppException(
            code="CLOUD_INTEGRATION_NOT_CONFIGURED",
            message=(
                f"integration {integration.slug or integration.id} has no encrypted "
                "credentials — re-upload provider credentials before DNS writes"
            ),
            status_code=409,
        )
    try:
        cred_fields = decrypt_fields(
            integration.secrets_encrypted, integration.secrets_nonce,
            integration_id=str(integration.id),
        )
    except CredentialDecryptionError as exc:
        raise AppException(
            code="CLOUD_INTEGRATION_CREDENTIAL_INVALID",
            message=(
                f"integration {integration.slug or integration.id} has invalid "
                f"credentials: {exc}"
            ),
            status_code=500,
        ) from exc
    # Merge public config (regions etc.) into the flat fields dict the
    # CloudProvider expects, matching the migration mapping.
    merged: dict[str, str] = {}
    if integration.config_public:
        merged.update({k: str(v) for k, v in integration.config_public.items() if v is not None})
    merged.update(cred_fields)
    creds = ProviderCredentials(
        provider_id=str(integration.id),
        provider_type=integration.provider,
        account_alias=integration.name,
        fields=merged,
    )
    return get_provider_class(integration.provider)(creds)


async def _assert_dns_writes_enabled(db: AsyncSession) -> None:
    """Enforce the ``cmdb.dns_writes_enabled`` freeze on ADMIN direct-write
    DNS endpoints too.

    The switch was only checked inside the self-service request-execute path
    (``DnsRequestService._execute_approved``), so a DNS "freeze" flipped in
    平台设置 during an incident still let any cloud_admin create/patch/delete
    records straight through ``/dns-records*`` — a partial freeze that
    contradicts the switch's own user-facing message. Gate every admin
    mutation on the same switch.
    """
    from app.domains.cmdb.prod_write_lock import (  # noqa: PLC0415
        DNS_LOCKED_MESSAGE,
        dns_writes_enabled,
    )
    if not await dns_writes_enabled(db):
        raise AppException(
            code="DNS_WRITES_LOCKED",
            message=DNS_LOCKED_MESSAGE,
            status_code=409,
        )


@router.post("/dns-records", status_code=201)
async def create_dns_record(
    body: DnsRecordCreate,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    await _assert_dns_writes_enabled(db)
    # ``body.integration_id`` is a ``uuid.UUID`` (Pydantic strict) and
    # every downstream consumer (``_load_cloud_integration`` does
    # ``uuid.UUID(integration_id)``; ``CmdbService.create_dns_record_in_db``
    # assigns to a ``String(36)`` column and the SELECT-WHERE binds against
    # the same column type — see the dns_request_service bug verified
    # on prod 2026-05-20 where ``CmdbDomain.integration_id == uuid.UUID``
    # returned 0 rows because asyncmy binds UUID as binary, not as the
    # hyphenated text MySQL stores). Cast at the boundary.
    integration_id_str = str(body.integration_id)
    provider = await _load_provider_for_dns(db, integration_id_str)
    try:
        dns_write = DnsRecordWrite(
            zone_name=body.zone_name,
            name=body.name,
            record_type=body.record_type,
            value=body.value,
            ttl=body.ttl,
            line=body.line,
        )
        provider_record_id = await provider.create_dns_record(dns_write)
    except Exception as exc:
        # exc_info=False: mirror the guard on
        # ``_set_dns_record_status_direct`` — Aliyun / Tencent SDK
        # stack frames sometimes carry the signed request URL (with
        # AccessKeyId= + signature) in their args. Logging the
        # traceback in full would leak provider credentials into ELK.
        logger.error("DNS create_dns_record failed: %s", exc)
        # Use AppException so the standard envelope wrapper produces
        # ``{success, data: null, meta, errors: [...]}`` — frontends
        # accessing ``response.data`` (even if null) won't crash.
        # Surface the redacted provider error so the toast is actionable
        # (e.g. "InvalidParameter.RR: ...") instead of opaque.
        from app.domains.cmdb.providers.exceptions import (
            redact_provider_error,
        )
        redacted = redact_provider_error(exc)
        raise AppException(
            code="CLOUD_DNS_WRITE_FAILED",
            message=(
                f"云服务商拒绝了 DNS 写入:{redacted}"
                if redacted
                else "Cloud provider DNS write failed (no detail returned)."
            ),
            status_code=502,
        ) from exc
    finally:
        await provider.close()

    # Round-28: admin direct-create must set owner + write history seed.
    # Previously the row landed with owner_person_id=NULL — every offboarding
    # scan, recertify cron, and "我的 DNS" view silently skipped it, and
    # there was no owner_history seed so the ownership timeline started
    # at NULL forever. Admin who supplies an explicit ``owner_person_id``
    # in the body wins; otherwise the admin themselves owns it (avoids
    # the silent-orphan failure mode). The explicit owner MUST be an
    # active employee — mirror the guard in `_resolve_owner_input` so
    # admins can't direct-create a record owned by a leaver_pending /
    # left person (which would silently orphan it once T+0 fires).
    if body.owner_person_id is not None:
        from app.identity.person_models import IdentityPerson
        person = await db.scalar(
            select(IdentityPerson).where(
                IdentityPerson.id == str(body.owner_person_id),
                IdentityPerson.deleted_at.is_(None),
            ),
        )
        if person is None:
            raise ValidationError(
                f"owner_person_id {body.owner_person_id!s} does not "
                "match any active person.",
            )
        if person.employment_status != "active":
            raise ValidationError(
                f"owner_person_id {body.owner_person_id!s} resolves to "
                f"{person.display_name or person.work_email or 'person'} "
                f"(employment_status={person.employment_status!r}); "
                "owner must be an active employee."
            )
    initial_owner = (
        str(body.owner_person_id) if body.owner_person_id else str(user.id)
    )
    try:
        record = await CmdbService.create_dns_record_in_db(
            db,
            integration_id=integration_id_str,
            zone_name=body.zone_name,
            name=body.name,
            record_type=body.record_type,
            value=body.value,
            ttl=body.ttl,
            line=body.line,
            provider_record_id=provider_record_id,
            owner_person_id=initial_owner,
            last_owner_review_at=datetime.now(UTC),
            actor_person_id=str(user.id),
        )
    except Exception as exc:
        # Cloud-side record is already created but the Helix DB insert
        # failed (FK error, deadlock, conflict). Log the orphan loudly
        # with everything an admin needs to manually delete the cloud
        # row — we don't try to auto-rollback the cloud delete from
        # inside an exception handler because each provider has
        # different idempotency semantics on delete-by-record-id and
        # a failed compensation could compound the orphan.
        # exc_info=False — provider SDK args may carry signed URLs.
        logger.error(
            "DNS create: ORPHAN CLOUD RECORD — provider %s zone %s "
            "created record_id=%s but Helix DB insert failed: %s",
            body.integration_id, body.zone_name, provider_record_id, exc,
        )
        raise AppException(
            code="CLOUD_DNS_ORPHAN_LOCAL_INSERT_FAILED",
            message=(
                "Cloud DNS record was created but local DB insert failed. "
                f"Admin must delete the cloud record manually "
                f"(provider record id: {provider_record_id})."
            ),
            status_code=500,
        ) from exc
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    # Round-28: write owner_history seed in the SAME transaction as the
    # row insert (helper now flushes only — caller commits below). The
    # previous split-commit pattern could leave a record with NULL
    # history if the second commit died.
    from app.domains.cmdb.owner_history import (
        CHANGE_REASON_ADMIN_CREATE,
        RESOURCE_TYPE_DNS_RECORD,
        record_owner_change,
    )
    await record_owner_change(
        db,
        resource_type=RESOURCE_TYPE_DNS_RECORD,
        resource_id=str(record.id),
        previous_owner_person_id=None,
        new_owner_person_id=initial_owner,
        actor_person_id=str(user.id),
        change_reason=CHANGE_REASON_ADMIN_CREATE,
        trace_id=trace_id,
        context={"provider_record_id": provider_record_id},
    )
    audit = AuditService(db)
    # Pre-launch hardening 2026-05-16: capture the full record state in
    # the audit row so forensic review of a malicious / mistaken DNS
    # change can reconstruct exactly what was written (target value, TTL,
    # line policy) without joining against ``cmdb_dns_records`` (which
    # may have been further mutated). DNS records are a high-impact
    # primitive — a wrong A/CNAME redirect can take down a service or
    # silently route traffic to an attacker.
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dns.create",
        status="success",
        resource_type="cmdb_dns_record",
        resource_id=str(record.id),
        detail={
            "zone_name": body.zone_name,
            "name": body.name,
            "type": body.record_type,
            # TXT values redacted to a hash prefix — see _audit_dns_value.
            "value": _audit_dns_value(body.record_type, body.value),
            "ttl": body.ttl,
            "line": body.line,
            "integration_id": str(body.integration_id),
            "provider_record_id": provider_record_id,
            "owner_person_id": initial_owner,
        },
    )
    await db.commit()
    await db.refresh(record)
    return success_response(data=DnsRecordResponse.model_validate(record).model_dump(), request=request)


@router.patch("/dns-records/{record_id}")
async def update_dns_record(
    record_id: uuid.UUID,
    body: DnsRecordUpdate,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    await _assert_dns_writes_enabled(db)
    from app.domains.cmdb.models import CmdbDnsRecord
    result = await db.execute(
        select(CmdbDnsRecord).where(
            CmdbDnsRecord.id == record_id,
            CmdbDnsRecord.deleted_at.is_(None),
        )
    )
    existing = result.scalar_one_or_none()
    if not existing:
        raise NotFoundError("DnsRecord", str(record_id))

    # Migration 173: enforce record_type → value format BEFORE we push
    # to the cloud. DnsRecordCreate validates this via model_validator,
    # but the update payload doesn't carry record_type (it's immutable
    # on the row) — we have to read the existing row's type and run the
    # same validator. Without this, an admin PATCH with a CNAME-shaped
    # value on an A record would get a 502 from the cloud provider's
    # opaque "InvalidParameter" rejection.
    #
    # Only validate when ``value`` actually changes — the edit modal
    # re-sends the existing value on every PATCH (initialValues), so
    # a CNAME row whose synced value is malformed (e.g. an IP slipped
    # in via a misclassified provider record) would otherwise reject
    # every owner-only / remark-only / ttl-only edit forever, even
    # though no value write is happening. Same root cause as the
    # has_content_change check below — "field present" ≠ "field
    # changed".
    if body.value is not None and body.value != existing.value:
        from app.domains.cmdb.schemas import _validate_dns_value
        try:
            _validate_dns_value(existing.record_type, body.value)
        except ValueError as exc:
            raise ValidationError(str(exc)) from exc

    # If actual content changed, push to cloud first.
    # Compare against the existing row — the edit modal pre-fills
    # name/value/ttl/line from the current record, so a pure owner-only
    # or remark-only edit still arrives with all four fields populated.
    # Using "is not None" alone treats every PATCH as a content change
    # and fires a redundant cloud write, which surfaces as "更新失败"
    # whenever the provider rejects the no-op (e.g. line-code mismatch
    # between display value and provider-accepted codes) — and at minimum
    # costs an unnecessary API call + ack on every owner reassignment.
    # ``body.name`` was dropped from DnsRecordUpdate (2026-05-28) — host
    # is now immutable on update. The change-detector keeps ``name`` out
    # of the comparison; ``DnsRecordWrite.name`` always sources from
    # ``existing.name``.
    has_content_change = (
        (body.value is not None and body.value != existing.value)
        or (body.ttl is not None and body.ttl != existing.ttl)
        or (body.line is not None and body.line != existing.line)
    )
    if has_content_change:
        provider = await _load_provider_for_dns(db, str(existing.integration_id))
        try:
            dns_write = DnsRecordWrite(
                zone_name=existing.zone_name,
                name=existing.name,
                record_type=existing.record_type,
                value=body.value or existing.value,
                ttl=body.ttl if body.ttl is not None else existing.ttl,
                line=body.line if body.line is not None else existing.line,
            )
            await provider.update_dns_record(existing.record_id, existing.zone_name, dns_write)
        except Exception as exc:
            # See create_dns_record above — exc_info=False to avoid
            # provider-SDK credential leak through tracebacks.
            logger.error("DNS update_dns_record failed: %s", exc)
            # Surface the redacted provider error to the caller so the
            # toast can show "InvalidParameter.RR: ..." instead of an
            # opaque "更新失败" the user can't act on. ``redact_provider_error``
            # strips AccessKey / STS / ARN / signed-URL fragments and
            # bounds length, so it's safe to send to a non-admin
            # frontend.
            from app.domains.cmdb.providers.exceptions import (
                redact_provider_error,
            )
            redacted = redact_provider_error(exc)
            raise AppException(
                code="CLOUD_DNS_WRITE_FAILED",
                message=(
                    f"云服务商拒绝了 DNS 写入:{redacted}"
                    if redacted
                    else "Cloud provider DNS write failed (no detail returned)."
                ),
                status_code=502,
            ) from exc
        finally:
            await provider.close()

    # Snapshot the pre-update state so the audit row records both
    # before + after — pre-launch hardening 2026-05-16. DNS is a high-
    # impact primitive (one wrong CNAME can hijack a service); we want
    # the audit trail to answer "what did the record look like before
    # this admin's edit?" without time-travelling against another
    # table.
    # Round-28: redact TXT values from audit before/after — see
    # _audit_dns_value rationale. Non-TXT pass through unchanged.
    before_state = {
        "name": existing.name,
        "value": _audit_dns_value(existing.record_type, existing.value),
        "ttl": existing.ttl,
        "line": existing.line,
    }
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    record = await CmdbService.update_dns_record_in_db(
        db, str(record_id), body, actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    if record is None:
        # Soft-deleted between the existing-fetch above and this update —
        # narrow race but real (a parallel admin delete could land in the
        # gap). Surface a clean 404 instead of crashing on
        # ``DnsRecordResponse.model_validate(None)``.
        raise NotFoundError("DnsRecord", str(record_id))
    audit = AuditService(db)
    after_state = {
        # host is immutable on update (schema-enforced) — always source
        # ``name`` from the existing row.
        "name": existing.name,
        "value": _audit_dns_value(
            existing.record_type,
            body.value if body.value is not None else existing.value,
        ),
        "ttl": body.ttl if body.ttl is not None else existing.ttl,
        "line": body.line if body.line is not None else existing.line,
    }
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dns.update",
        status="success",
        resource_type="cmdb_dns_record",
        resource_id=str(record_id),
        detail={
            "zone_name": existing.zone_name,
            "type": existing.record_type,
            "integration_id": str(existing.integration_id),
            "provider_record_id": existing.record_id,
            "before": before_state,
            "after": after_state,
            # Round-28: scrub TXT value from the raw patch payload so the
            # audit JSON doesn't leak the original (the before/after slots
            # are already redacted via _audit_dns_value).
            "patch": _scrub_audit_patch(
                existing.record_type,
                body.model_dump(exclude_unset=True, mode="json"),
            ),
        },
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [record])
    item = _augment_response(
        DnsRecordResponse.model_validate(record).model_dump(), record, labels, owner_meta,
    )
    return success_response(data=item, request=request)


@router.delete("/dns-records/{record_id}", status_code=204)
async def delete_dns_record(
    record_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    await _assert_dns_writes_enabled(db)
    from app.domains.cmdb.models import CmdbDnsRecord
    # Lock the row so concurrent double-submits can't both fire the
    # irreversible provider delete. The second request blocks, re-reads
    # after the first commits the soft-delete, and falls through to a
    # clean NotFound rather than a second cloud-side delete.
    result = await db.execute(
        select(CmdbDnsRecord).where(
            CmdbDnsRecord.id == record_id,
            CmdbDnsRecord.deleted_at.is_(None),
        ).with_for_update()
    )
    existing = result.scalar_one_or_none()
    if not existing:
        raise NotFoundError("DnsRecord", str(record_id))

    provider = await _load_provider_for_dns(db, str(existing.integration_id))
    try:
        await provider.delete_dns_record(existing.record_id, existing.zone_name)
    except Exception as exc:
        # See create_dns_record above — exc_info=False to avoid
        # provider-SDK credential leak through tracebacks.
        logger.error("DNS delete_dns_record failed: %s", exc)
        from app.domains.cmdb.providers.exceptions import (
            redact_provider_error,
        )
        redacted = redact_provider_error(exc)
        raise AppException(
            code="CLOUD_DNS_WRITE_FAILED",
            message=(
                f"云服务商拒绝了 DNS 删除:{redacted}"
                if redacted
                else "Cloud provider DNS delete failed (no detail returned)."
            ),
            status_code=502,
        ) from exc
    finally:
        await provider.close()

    # Capture the full deleted record in the audit detail so an
    # admin can reconstruct the lost row (zone+name+type+value+ttl+
    # line) if a deletion was mistaken or malicious — pre-launch
    # hardening 2026-05-16. The cloud-side delete already returned
    # ok above; without this snapshot the only forensic source is
    # the (now soft-deleted) ``cmdb_dns_records`` row.
    previous_owner = existing.owner_person_id
    await CmdbService.delete_dns_record_in_db(
        db, str(record_id), actor_person_id=str(user.id),
    )
    # Round-28: write an owner_history TERMINAL row so the timeline
    # admin UI shows the record's lifecycle ending here. Same
    # transaction as the soft-delete (helper now only flushes).
    from app.domains.cmdb.owner_history import (
        CHANGE_REASON_ADMIN_PATCH,
        RESOURCE_TYPE_DNS_RECORD,
        record_owner_change,
    )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    if previous_owner is not None:
        await record_owner_change(
            db,
            resource_type=RESOURCE_TYPE_DNS_RECORD,
            resource_id=str(record_id),
            previous_owner_person_id=str(previous_owner),
            new_owner_person_id=None,
            actor_person_id=str(user.id),
            change_reason=CHANGE_REASON_ADMIN_PATCH,
            trace_id=trace_id,
            context={"action": "admin_delete"},
        )
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dns.delete",
        status="success",
        resource_type="cmdb_dns_record",
        resource_id=str(record_id),
        detail={
            "zone_name": existing.zone_name,
            "name": existing.name,
            "type": existing.record_type,
            # Round-28: redact TXT values from audit detail — TXT
            # records carry domain-ownership tokens / ACME challenges
            # / SaaS verification secrets that should not be plain-
            # text in audit_logs.detail (readable by every auditor).
            # Hash to a 16-char prefix so forensic compare-by-hash
            # still works.
            "value": _audit_dns_value(existing.record_type, existing.value),
            "ttl": existing.ttl,
            "line": existing.line,
            "integration_id": str(existing.integration_id),
            "provider_record_id": existing.record_id,
        },
    )
    await db.commit()


# ---------------------------------------------------------------------------
# DNS admin direct-action endpoints (migration 173)
#
# Admin can flip a record's enable/disable status without going through
# the Feishu approval flow; employees still go through the request flow
# (DnsRecordRequest with action=enable|disable). Both paths land in the
# same end state (cloud-side status flipped + cmdb_dns_records.status
# updated + audit log).
# ---------------------------------------------------------------------------


async def _set_dns_record_status_direct(
    *,
    record_id: uuid.UUID,
    request: Request,
    user: AuthUser,
    db: AsyncSession,
    enable: bool,
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    await _assert_dns_writes_enabled(db)
    from app.domains.cmdb.models import CmdbDnsRecord
    from app.domains.cmdb.providers.exceptions import CapabilityNotSupported
    result = await db.execute(
        select(CmdbDnsRecord).where(
            CmdbDnsRecord.id == record_id,
            CmdbDnsRecord.deleted_at.is_(None),
        )
    )
    existing = result.scalar_one_or_none()
    if not existing:
        raise NotFoundError("DnsRecord", str(record_id))
    desired = "enable" if enable else "disable"
    if existing.status == desired:
        # No-op — return the row unchanged. The cloud may have been
        # flipped externally; we don't second-guess unless we observe
        # a delta from the cmdb_dns_records.status snapshot. We still
        # audit the click so "who pressed the button" survives in the
        # log even when the system did nothing (DNS is high-impact
        # governance; silent successes are worse than noisy ones).
        trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
        audit = AuditService(db)
        await audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=user.id,
            action=f"cmdb.dns.{'enable' if enable else 'disable'}",
            status="noop",
            resource_type="cmdb_dns_record",
            resource_id=str(record_id),
            detail={
                "zone_name": existing.zone_name,
                "name": existing.name,
                "current_status": existing.status,
                "reason": "already_in_target_state",
            },
        )
        await db.commit()
        labels, owner_meta = await _resolve_labels_and_owner_meta(db, [existing])
        item = _augment_response(
            DnsRecordResponse.model_validate(existing).model_dump(),
            existing, labels, owner_meta,
        )
        return success_response(data=item, request=request)

    provider = await _load_provider_for_dns(db, str(existing.integration_id))
    try:
        await provider.set_dns_record_status(
            existing.record_id, existing.zone_name, enable,
        )
    except CapabilityNotSupported as exc:
        raise AppException(
            code="DNS_PROVIDER_NO_ENABLE_DISABLE",
            message=str(exc),
            status_code=501,
        ) from exc
    except Exception as exc:
        # exc_info=False: Aliyun / Tencent SDK stack frames sometimes
        # carry the signed request URL (with ``AccessKeyId=`` and the
        # signature) in their ``args``. Logging the traceback in full
        # would leak credentials into ELK. The exception message itself
        # has already been run through ``redact_provider_error`` /
        # ``_translate_sdk_exc`` at the driver boundary.
        logger.error("DNS set_dns_record_status failed: %s", exc)
        from app.domains.cmdb.providers.exceptions import (
            redact_provider_error,
        )
        redacted = redact_provider_error(exc)
        raise AppException(
            code="CLOUD_DNS_WRITE_FAILED",
            message=(
                f"云服务商拒绝了 DNS 状态切换:{redacted}"
                if redacted
                else "Cloud provider DNS status flip failed (no detail returned)."
            ),
            status_code=502,
        ) from exc
    finally:
        await provider.close()

    before_status = existing.status
    record = await CmdbService.set_dns_record_status_in_db(
        db, str(record_id), enable=enable, actor_person_id=str(user.id),
    )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action=f"cmdb.dns.{'enable' if enable else 'disable'}",
        status="success",
        resource_type="cmdb_dns_record",
        resource_id=str(record_id),
        detail={
            "zone_name": existing.zone_name,
            "name": existing.name,
            "type": existing.record_type,
            "integration_id": str(existing.integration_id),
            "provider_record_id": existing.record_id,
            "before_status": before_status,
            "after_status": desired,
        },
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [record])
    item = _augment_response(
        DnsRecordResponse.model_validate(record).model_dump(), record, labels, owner_meta,
    )
    return success_response(data=item, request=request)


@router.post("/dns-records/{record_id}/enable")
async def enable_dns_record(
    record_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Admin-only direct flip of a record's cloud-side status to ENABLE.

    Employees should submit a DnsRecordRequest with action='enable'
    instead — that goes through Feishu approval.
    """
    return await _set_dns_record_status_direct(
        record_id=record_id, request=request, user=user, db=db, enable=True,
    )


@router.post("/dns-records/{record_id}/disable")
async def disable_dns_record(
    record_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Admin-only direct flip of a record's cloud-side status to DISABLE."""
    return await _set_dns_record_status_direct(
        record_id=record_id, request=request, user=user, db=db, enable=False,
    )


@router.post("/dns-records/batch-owner-transfer")
async def batch_transfer_dns_record_owner(
    body: DnsRecordBatchOwnerTransfer,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Self-service DNS owner handoff.

    Active employee picks 1-500 DNS records they own and a new active
    owner. Service validates ownership of EVERY selected record (admin
    bypass) and reassigns them atomically — no approval, since the
    caller is not gaining any new privilege by handing off rows they
    already control. New owner gets a Feishu + email notification via
    the outbox so they know they have inherited the records.

    Distinct from the admin-only ``/dns-records/batch-owner`` endpoint:
    that one accepts any record-id (admin grants itself the cross-fleet
    write) and supports clearing the owner to null. Self-service can do
    neither.
    """
    if not _can_submit_self_service_dns(user.role_names):
        raise ForbiddenError()
    is_admin = can_manage_cloud(user.role_names)
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    summary = await CmdbService.transfer_owned_dns_records(
        db,
        record_ids=[str(rid) for rid in body.record_ids],
        new_owner_person_id=str(body.new_owner_person_id),
        caller_person_id=str(user.id),
        caller_is_admin=is_admin,
        reason=body.reason,
        trace_id=trace_id,
    )
    if summary["not_owned"] and not is_admin:
        # Non-admin caller listed records they don't own. Surface as
        # 403 with the count so the FE can render "你不是这 N 条记录的
        # owner, 无法转移" and re-fetch the page.
        raise ForbiddenError(
            f"{summary['not_owned']} record(s) in the batch are not "
            "owned by you. Refresh the list and try again."
        )
    transferred_ids = summary.pop("_transferred_ids", [])
    audit = AuditService(db)
    # ``audit_logs.resource_id`` is String(128); ",".join of 50 UUIDs is
    # ~1849 chars and silently truncates to ~3 entries, breaking
    # forensic search by exact id. Store the first id as the row key
    # and the full list in detail so a ``WHERE resource_id = '...'``
    # query lands at least one match while the full set stays available
    # via ``detail.transferred_ids`` for ad-hoc analytics.
    first_id = transferred_ids[0] if transferred_ids else ""
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dns.self_service_transfer",
        status="success",
        resource_type="cmdb_dns_record",
        resource_id=first_id,
        detail={
            "summary": summary,
            "new_owner_person_id": str(body.new_owner_person_id),
            "record_ids_count": len(body.record_ids),
            "transferred_ids": transferred_ids,
            "reason": body.reason,
        },
    )
    # Best-effort notify the new owner via outbox. Failure to dispatch
    # never blocks the transfer — the rows have already committed; the
    # notification merely tells the new owner what happened.
    if transferred_ids:
        try:
            await _notify_owner_of_transfer(
                db,
                new_owner_person_id=str(body.new_owner_person_id),
                actor=user,
                resource_kind="dns",
                resource_count=len(transferred_ids),
                reason=body.reason,
                trace_id=trace_id,
            )
        except Exception:  # noqa: BLE001
            logger.exception(
                "DNS self-service transfer: notify dispatch failed "
                "caller=%s new_owner=%s count=%d",
                user.id, body.new_owner_person_id, len(transferred_ids),
            )
    return success_response(data=summary, request=request)


@router.post("/dns-records/batch-owner")
async def batch_update_dns_record_owner(
    body: DnsRecordBatchOwnerUpdate,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Bulk-reassign DNS record owners (admin only).

    Mirrors the legacy CMDB's batch owner update mode of
    ``PATCH /zones/.../records`` (when ``record_ids[]`` was populated
    and the only changed field was ``owner``). Pure DB-side write —
    cloud DNS providers don't know about Helix's owner FK.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    # Round-28: hoist trace_id ABOVE the service call so owner_history
    # rows for each batched record carry the same trace_id as the
    # surrounding audit log row (forensic correlation across the two
    # tables). Previously the trace_id was generated AFTER the service
    # call, leaving every history row with NULL trace_id.
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    summary = await CmdbService.batch_update_dns_record_owner(
        db,
        [str(rid) for rid in body.record_ids],
        str(body.owner_person_id) if body.owner_person_id else None,
        actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dns.batch_owner_update",
        status="success",
        resource_type="cmdb_dns_record",
        resource_id=",".join(str(rid) for rid in body.record_ids[:50]),
        detail={
            "summary": summary,
            "new_owner_person_id": str(body.owner_person_id) if body.owner_person_id else None,
            "record_ids_count": len(body.record_ids),
        },
    )
    return success_response(data=summary, request=request)


# ---------------------------------------------------------------------------
# DNS self-service requests (employee-facing approval flow)
# ---------------------------------------------------------------------------


@router.post("/dns-record-requests", status_code=201)
async def submit_dns_record_request(
    body: DnsRecordRequestCreate,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_approval_submit_rate_limit),
):
    """Employee submits a create / update / delete DNS request.

    Round-28: gated to ``employee`` + ``global_admin`` only. Pure
    read-only roles (auditor / management / cloud_admin-without-
    employee) can no longer reach the Feishu approval flow. Ownership
    of the target record is still enforced at the body level inside
    ``DnsRequestService.submit``.
    """
    if not _can_submit_self_service_dns(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.dns_request_service import DnsRequestService

    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    # Round-49: use get_real_ip so X-Forwarded-For (nginx + Docker)
    # lands in the audit row instead of 127.0.0.1.
    ip = get_real_ip(request)
    ua = request.headers.get("user-agent")
    svc = DnsRequestService(db)
    response = await svc.submit(
        body=body,
        requester=user,
        trace_id=trace_id,
        ip_address=ip,
        user_agent=ua,
    )
    await db.commit()
    return success_response(data=response.model_dump(), request=request)


@router.get("/dns-record-requests")
async def list_dns_record_requests(
    request: Request,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    status: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List the caller's own DNS requests.

    Admins still see only their own here — the admin-facing centre
    is /api/v1/approval/admin/requests which already lists every
    approval type uniformly.
    """
    from app.domains.cmdb.dns_request_service import DnsRequestService

    svc = DnsRequestService(db)
    items, total = await svc.list_for_user(
        requester=user, page=page, page_size=page_size, status=status,
    )
    data = [item.model_dump() for item in items]
    return paginated_response(
        data=data, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/dns-record-requests/{request_id}")
async def get_dns_record_request(
    request_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.cmdb.dns_request_service import DnsRequestService

    svc = DnsRequestService(db)
    item = await svc.get(str(request_id), requester=user)
    return success_response(data=item.model_dump(), request=request)


@router.post("/dns-record-requests/{request_id}/cancel")
async def cancel_dns_record_request(
    request_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.cmdb.dns_request_service import DnsRequestService

    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    svc = DnsRequestService(db)
    item = await svc.cancel(
        request_id=str(request_id), requester=user, trace_id=trace_id,
    )
    await db.commit()
    # False ⇒ Helix cancelled but the parallel Feishu instance is still
    # pending (e.g. template withdraw disabled) — frontend warns.
    return success_response(
        data={
            **item.model_dump(),
            "feishu_cancel_synced": svc.last_feishu_cancel_synced,
        },
        request=request,
    )


class _DnsRequestBypassBody(BaseModel):
    """Body for ``POST /dns-record-requests/{id}/bypass-execute``.

    ``reason`` is the admin's written attestation of why bypassing the
    4-eyes guard is justified. It lands verbatim in the audit log + in
    every other admin's notification card, so the field is treated as a
    human-readable sentence (not an enum slug).
    """

    model_config = {"extra": "forbid"}
    reason: str = Field(min_length=20, max_length=2000)


@router.post("/dns-record-requests/{request_id}/bypass-execute")
async def bypass_4_eyes_dns_request(
    request_id: uuid.UUID,
    body: _DnsRequestBypassBody,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    # Pre-launch P0 fix (Round 45 follow-up, CR-208): browser-session
    # gate on the most-destructive cloud-write override. A stolen
    # cloud_admin PAT must not be sufficient to fire the 4-eyes
    # bypass; an interactive web session is required. MCP tool
    # ``bypass_4_eyes_dns_request`` keeps working (calls the service
    # directly, not this HTTP endpoint). Declared AFTER
    # ``get_current_user`` so ``request.state.auth_via`` is set
    # before the gate fail-closes on its missing-default branch.
    _gate: None = Depends(require_browser_session),
    db: AsyncSession = Depends(get_db),
):
    """Admin override of a 4-eyes-blocked DNS request.

    Service-side gates:
      * ``can_manage_cloud`` (global_admin / cloud_admin) — auditor /
        management / employee → 403.
      * row must be in ``failed`` state with an error_message that
        carries the 4-eyes fragment → 409 otherwise.
      * ``reason`` must be ≥20 chars (enforced again at the body schema).

    Every successful bypass:
      * audit-logs ``cmdb.dns_request.bypassed_4_eyes_execute`` with the
        reason + original error_message
      * fans out a Feishu + email notification to EVERY other admin
        (global_admin / cloud_admin / it_admin)
      * proceeds to the cloud DNS write (the actual provider call) —
        same execute path as a normal approval.
    """
    from app.domains.cmdb.dns_request_service import DnsRequestService

    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    # Round-49: get_real_ip (X-Forwarded-For aware) — see comment on
    # ``submit_dns_request`` for rationale.
    ip = get_real_ip(request)
    ua = request.headers.get("user-agent")
    svc = DnsRequestService(db)
    item = await svc.bypass_4_eyes_execute(
        request_id=str(request_id),
        actor=user,
        reason=body.reason,
        trace_id=trace_id,
        ip_address=ip,
        user_agent=ua,
    )
    # bypass_4_eyes_execute commits internally before the cloud write so
    # the approved-marker + audit + notifications are durable. The post-
    # cloud-write commit is also internal. Nothing left to commit here.
    return success_response(data=item.model_dump(), request=request)


@router.post("/dns-record-requests/{request_id}/retry")
async def retry_dns_request(
    request_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Re-open a fresh Feishu approval instance for a 4-eyes-blocked
    request, after the admin has fixed the template's approver routing.

    No bypass — the new instance goes through Feishu's approver chain
    as normal, and the server-side 4-eyes guard will fire again on the
    new callback if the template still routes back to the requester.
    """
    from app.domains.cmdb.dns_request_service import DnsRequestService

    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    # Round-49: get_real_ip — see ``submit_dns_request`` comment.
    ip = get_real_ip(request)
    ua = request.headers.get("user-agent")
    svc = DnsRequestService(db)
    item = await svc.retry_after_template_fix(
        request_id=str(request_id),
        actor=user,
        trace_id=trace_id,
        ip_address=ip,
        user_agent=ua,
    )
    await db.commit()
    return success_response(data=item.model_dump(), request=request)


@router.post("/dns-record-requests/{request_id}/resume")
async def resume_dns_request(
    request_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Resume a DNS request that failed ONLY because DNS writes were
    locked at execute time (``cmdb.dns_writes_enabled`` / the legacy
    prod-cloud freeze). The approval is still valid, so this re-drives
    the cloud write on the same instance — no new approval, no bypass.

    Requires ``cloud_admin`` / ``global_admin``. The service refuses if
    the switch is still locked or the failure was a genuine cloud-write
    error (not a lock).
    """
    from app.domains.cmdb.dns_request_service import DnsRequestService

    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    ip = get_real_ip(request)
    ua = request.headers.get("user-agent")
    svc = DnsRequestService(db)
    item = await svc.resume_after_write_lock_cleared(
        request_id=str(request_id),
        actor=user,
        trace_id=trace_id,
        ip_address=ip,
        user_agent=ua,
    )
    await db.commit()
    return success_response(data=item.model_dump(), request=request)


@router.post("/dns-records/recertify")
async def recertify_dns_records(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Owner clicks "all still in use" on the recertify Feishu card or
    a button in /my/cloud/dns. Body shape: ``{record_ids?: [uuid]}`` —
    omit to recertify EVERY record the owner currently owns.

    Round-28: gated to ``employee`` + ``global_admin`` only. Same
    rationale as ``POST /dns-record-requests`` — read-only privileged
    roles should not be re-stamping recertification timestamps. The
    service-side filter (``WHERE owner_person_id == self.id``) already
    ensures the request is a no-op for non-owners, but the row would
    still produce a noisy ``cmdb.dns.recertify`` audit entry with
    count=0 — drop that surface upfront.
    """
    if not _can_submit_self_service_dns(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.dns_request_service import DnsRequestService

    # Treat malformed JSON as a 400 instead of silently coercing to
    # ``{}`` (which used to mean "recertify everything I own"). A
    # caller asking for "all" must explicitly send ``{}`` or no body
    # via Content-Length: 0. Anything else is a client error.
    body_raw = await request.body()
    if body_raw:
        import json as _json
        try:
            body_json = _json.loads(body_raw)
        except _json.JSONDecodeError as exc:
            raise ValidationError(
                f"Malformed JSON body: {exc.msg}"
            ) from exc
        if not isinstance(body_json, dict):
            raise ValidationError("Request body must be a JSON object.")
    else:
        body_json = {}
    record_ids = body_json.get("record_ids")
    if record_ids is not None and not isinstance(record_ids, list):
        raise ValidationError("record_ids must be a list of uuids or omitted.")
    if isinstance(record_ids, list) and len(record_ids) > 1000:
        raise ValidationError(
            "record_ids list capped at 1000 — split into multiple calls "
            "or omit record_ids to recertify every record you own."
        )
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    svc = DnsRequestService(db)
    requested = (
        len(record_ids) if isinstance(record_ids, list) else None
    )
    count = await svc.recertify_owned(
        owner=user,
        record_ids=record_ids,
        trace_id=trace_id,
    )
    await db.commit()
    response_payload: dict = {"recertified": count}
    if requested is not None:
        # Surface the gap so the caller (and the audit log row) can
        # distinguish "user requested 50, owns 50" from "user requested
        # 50, owns 12" — important when the recertify came from a
        # Feishu card that pre-listed records the user used to own.
        response_payload["requested"] = requested
        response_payload["skipped"] = max(requested - count, 0)
    return success_response(data=response_payload, request=request)


# ---------------------------------------------------------------------------
# Load Balancers
# ---------------------------------------------------------------------------


@router.get("/load-balancers")
async def list_load_balancers(
    request: Request,
    integration_id: str | None = None,
    region: str | None = None,
    network_type: str | None = None,
    search: str | None = None,
    owner_person_id: str | None = None,
    page: int = 1,
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Migration 173 added ``owner_person_id`` FK to LBs — they now follow
    # the same view-cloud-resources gate as servers/domains/dns/cdn and
    # the same per-user / per-manager scope demotion.
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    owner_person_ids: list[str] | None = None
    if not can_read_cloud(user.role_names):
        subs = await get_subordinate_person_ids(db, str(user.id))
        if subs:
            owner_person_ids = [str(user.id), *subs]
            owner_person_id = None
        else:
            owner_person_id = str(user.id)
    lbs, total = await CmdbService.list_load_balancers(
        db,
        integration_id=integration_id,
        region=region,
        network_type=network_type,
        search=search,
        owner_person_id=owner_person_id,
        owner_person_ids=owner_person_ids,
        page=page,
        page_size=page_size,
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, lbs)
    data = [
        _augment_response(LoadBalancerResponse.model_validate(lb).model_dump(), lb, labels, owner_meta)
        for lb in lbs
    ]
    return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request)


@router.get("/load-balancers/{lb_id}")
async def get_load_balancer(
    lb_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbLoadBalancer
    result = await db.execute(
        select(CmdbLoadBalancer).where(
            CmdbLoadBalancer.id == lb_id,
            CmdbLoadBalancer.deleted_at.is_(None),
        )
    )
    lb = result.scalar_one_or_none()
    if not lb:
        raise NotFoundError("LoadBalancer", str(lb_id))
    # Per-row scope check: owners/managers/admins only.
    if not can_read_cloud(user.role_names):
        is_owner = lb.owner_person_id == str(user.id)
        if not is_owner:
            subs = set(await get_subordinate_person_ids(db, str(user.id)))
            if not lb.owner_person_id or lb.owner_person_id not in subs:
                raise NotFoundError("LoadBalancer", str(lb_id))
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [lb])
    item = _augment_response(
        LoadBalancerResponse.model_validate(lb).model_dump(), lb, labels, owner_meta,
    )
    return success_response(data=item, request=request)


@router.patch("/load-balancers/{lb_id}")
async def annotate_load_balancer(
    lb_id: uuid.UUID,
    body: LoadBalancerAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbLoadBalancer as _CmdbLb  # noqa: PLC0415
    pre = await db.scalar(
        select(_CmdbLb).where(
            _CmdbLb.id == lb_id,
            _CmdbLb.deleted_at.is_(None),
        ),
    )
    before_owner = pre.owner_person_id if pre else None
    before_owner_raw = pre.owner_raw if pre else None
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    lb = await CmdbService.annotate_load_balancer(
        db, str(lb_id), body, actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    if not lb:
        raise NotFoundError("LoadBalancer", str(lb_id))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.load_balancer.annotate",
        status="success",
        resource_type="cmdb_load_balancer",
        resource_id=str(lb_id),
        detail={
            "patch": body.model_dump(exclude_unset=True, mode="json"),
            "before": {
                "owner_person_id": before_owner,
                "owner_raw": before_owner_raw,
            },
            "after": {
                "owner_person_id": lb.owner_person_id,
                "owner_raw": lb.owner_raw,
            },
        },
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [lb])
    item = _augment_response(
        LoadBalancerResponse.model_validate(lb).model_dump(), lb, labels, owner_meta,
    )
    return success_response(data=item, request=request)


@router.post("/load-balancers/batch-annotate")
async def batch_annotate_load_balancers(
    body: LoadBalancerBatchAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Apply the same annotation to multiple LBs in one round-trip."""
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    # Round-28: hoist trace_id ABOVE the service call so per-row
    # owner_history rows carry the same trace_id as the audit log row.
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    summary = await CmdbService.batch_annotate_load_balancers(
        db,
        [str(lid) for lid in body.load_balancer_ids],
        body.annotation,
        actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.load_balancer.batch_annotate",
        status="success",
        resource_type="cmdb_load_balancer",
        resource_id=",".join(str(s) for s in body.load_balancer_ids[:50]),
        detail={
            "summary": summary,
            "annotation": body.annotation.model_dump(exclude_unset=True, mode="json"),
            "load_balancer_ids_count": len(body.load_balancer_ids),
        },
    )
    return success_response(data=summary, request=request)


# ---------------------------------------------------------------------------
# CDN Domains
# ---------------------------------------------------------------------------


@router.get("/cdn-domains")
async def list_cdn_domains(
    request: Request,
    integration_id: str | None = None,
    status: str | None = None,
    search: str | None = None,
    owner_person_id: str | None = None,
    page: int = 1,
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    owner_person_ids: list[str] | None = None
    if not can_read_cloud(user.role_names):
        subs = await get_subordinate_person_ids(db, str(user.id))
        if subs:
            owner_person_ids = [str(user.id), *subs]
            owner_person_id = None
        else:
            owner_person_id = str(user.id)
    domains, total = await CmdbService.list_cdn_domains(
        db,
        integration_id=integration_id,
        status=status,
        search=search,
        owner_person_id=owner_person_id,
        owner_person_ids=owner_person_ids,
        page=page,
        page_size=page_size,
    )
    owner_labels = await _resolve_owner_labels(
        db, [d.owner_person_id for d in domains if d.owner_person_id],
    )
    owner_meta = await _resolve_owner_meta(
        db, [d.owner_person_id for d in domains if d.owner_person_id],
    )
    data = []
    for d in domains:
        item = CdnDomainResponse.model_validate(d).model_dump()
        if d.owner_person_id:
            item["owner_label"] = owner_labels.get(d.owner_person_id)
            m = owner_meta.get(d.owner_person_id)
            if m:
                item["owner_employee_no"] = m["employee_no"]
                item["owner_employment_status"] = m["employment_status"]
        data.append(item)
    return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request)


@router.get("/cdn-domains/{domain_id}")
async def get_cdn_domain(
    domain_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbCdnDomain
    result = await db.execute(
        select(CmdbCdnDomain).where(
            CmdbCdnDomain.id == domain_id,
            CmdbCdnDomain.deleted_at.is_(None),
        )
    )
    domain = result.scalar_one_or_none()
    if not domain:
        raise NotFoundError("CdnDomain", str(domain_id))
    # Per-row ownership check mirrors list endpoint: owner OR manager OR admin.
    if not can_read_cloud(user.role_names):
        is_owner = domain.owner_person_id == str(user.id)
        if not is_owner:
            subs = set(await get_subordinate_person_ids(db, str(user.id)))
            if not domain.owner_person_id or domain.owner_person_id not in subs:
                raise NotFoundError("CdnDomain", str(domain_id))
    item = CdnDomainResponse.model_validate(domain).model_dump()
    if domain.owner_person_id:
        labels = await _resolve_owner_labels(db, [domain.owner_person_id])
        item["owner_label"] = labels.get(domain.owner_person_id)
        owner_meta = await _resolve_owner_meta(db, [domain.owner_person_id])
        m = owner_meta.get(str(domain.owner_person_id))
        if m:
            item["owner_employee_no"] = m["employee_no"]
            item["owner_employment_status"] = m["employment_status"]
    return success_response(data=item, request=request)


@router.patch("/cdn-domains/{cdn_id}")
async def annotate_cdn_domain(
    cdn_id: uuid.UUID,
    body: CdnDomainAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbCdnDomain as _CmdbCdnDomain  # noqa: PLC0415
    pre = await db.scalar(
        select(_CmdbCdnDomain).where(
            _CmdbCdnDomain.id == cdn_id,
            _CmdbCdnDomain.deleted_at.is_(None),
        ),
    )
    before_owner = pre.owner_person_id if pre else None
    before_owner_raw = pre.owner_raw if pre else None
    before_cert = pre.cert_id if pre else None
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    cdn = await CmdbService.annotate_cdn_domain(
        db, str(cdn_id), body, actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    if not cdn:
        raise NotFoundError("CdnDomain", str(cdn_id))
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.cdn_domain.annotate",
        status="success",
        resource_type="cmdb_cdn_domain",
        resource_id=str(cdn_id),
        detail={
            "patch": body.model_dump(exclude_unset=True, mode="json"),
            "before": {
                "owner_person_id": before_owner,
                "owner_raw": before_owner_raw,
                "cert_id": before_cert,
            },
            "after": {
                "owner_person_id": cdn.owner_person_id,
                "owner_raw": cdn.owner_raw,
                "cert_id": cdn.cert_id,
            },
        },
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [cdn])
    item = _augment_response(
        CdnDomainResponse.model_validate(cdn).model_dump(), cdn, labels, owner_meta,
    )
    return success_response(data=item, request=request)


# ---------------------------------------------------------------------------
# Object Storage
# ---------------------------------------------------------------------------


@router.get("/object-storage")
async def list_object_storage(
    request: Request,
    integration_id: str | None = None,
    region: str | None = None,
    status: str | None = None,
    search: str | None = None,
    owner_person_id: str | None = None,
    unowned: bool = False,
    page: int = 1,
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Gate on can_read_cloud — admin/auditor/finance scope only. Bucket
    # names + ACL state (especially public-read flags) are
    # reconnaissance gold; never expose to plain employees. (Migration
    # 366 did give OSS an ``owner_person_id``, but unlike servers / LBs
    # this list is NOT demoted to a self/subordinate view for readers
    # outside that gate — they don't reach it at all.)
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    buckets, total = await CmdbService.list_object_storage(
        db,
        integration_id=integration_id,
        region=region,
        status=status,
        search=search,
        owner_person_id=owner_person_id,
        unowned=unowned,
        page=page,
        page_size=page_size,
    )
    # Migration 366: OSS carries owner_person_id — resolve owner_label +
    # leaver-meta at serialize time, same as the load-balancer list.
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, buckets)
    data = [
        _augment_response(ObjectStorageResponse.model_validate(b).model_dump(), b, labels, owner_meta)
        for b in buckets
    ]
    return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request)


@router.get("/object-storage/{oss_id}")
async def get_object_storage(
    oss_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Single object-storage bucket by id. Same ``can_read_cloud`` gate
    as the list (OSS rows carry no owner — bucket names + public-read
    ACL flags are admin/auditor-only recon surface). Added so the mobile
    object-storage detail deep-link resolves by id instead of paging the
    list and ``.find()``-ing (silent 404 past the first page)."""
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbObjectStorage
    row = (await db.execute(
        select(CmdbObjectStorage).where(
            CmdbObjectStorage.id == oss_id,
            CmdbObjectStorage.deleted_at.is_(None),
        ),
    )).scalar_one_or_none()
    if row is None:
        raise NotFoundError("ObjectStorage", str(oss_id))
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [row])
    item = _augment_response(
        ObjectStorageResponse.model_validate(row).model_dump(), row, labels, owner_meta,
    )
    return success_response(data=item, request=request)


@router.patch("/object-storage/{oss_id}")
async def annotate_object_storage(
    oss_id: uuid.UUID,
    body: ObjectStorageAnnotation,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbObjectStorage as _CmdbOss  # noqa: PLC0415
    pre = await db.scalar(
        select(_CmdbOss).where(
            _CmdbOss.id == oss_id,
            _CmdbOss.deleted_at.is_(None),
        ),
    )
    before_owner = pre.owner_person_id if pre else None
    before_owner_raw = pre.owner_raw if pre else None
    trace_id = getattr(request.state, "trace_id", str(uuid.uuid4()))
    oss = await CmdbService.annotate_object_storage(
        db, str(oss_id), body,
        actor_person_id=str(user.id),
        trace_id=trace_id,
    )
    if not oss:
        raise NotFoundError("ObjectStorage", str(oss_id))
    # State-changing PATCH carries an audit trail. Mirror the cdn_domain /
    # load_balancer annotate audit shape (before/after snapshot + patch
    # payload) so the audit_logs surface is uniform across every network /
    # cloud annotate endpoint. Migration 366 added owner_person_id, so the
    # snapshot now tracks both the FK and the cloud-tag fallback.
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.object_storage.annotate",
        status="success",
        resource_type="cmdb_object_storage",
        resource_id=str(oss_id),
        detail={
            "patch": body.model_dump(exclude_unset=True, mode="json"),
            "before": {
                "owner_person_id": before_owner,
                "owner_raw": before_owner_raw,
            },
            "after": {
                "owner_person_id": oss.owner_person_id,
                "owner_raw": oss.owner_raw,
            },
        },
    )
    labels, owner_meta = await _resolve_labels_and_owner_meta(db, [oss])
    item = _augment_response(
        ObjectStorageResponse.model_validate(oss).model_dump(), oss, labels, owner_meta,
    )
    return success_response(data=item, request=request)


# ---------------------------------------------------------------------------
# IAM users
# ---------------------------------------------------------------------------


@router.get("/iam-users")
async def list_iam_users(
    request: Request,
    integration_id: str | None = None,
    search: str | None = None,
    mfa: str | None = Query(default=None, pattern="^(true|false)?$"),
    has_owner: str | None = Query(
        default=None, pattern="^(true|false)?$",
        description=(
            "Owner-mapping filter. ``true`` → only IAM users mapped to "
            "a Helix person; ``false`` → orphan IAM users: no owner AND "
            "not classified as a service account (compliance review "
            "surface — a declared robot is not an orphan)."
        ),
    ),
    account_kind: str | None = Query(
        default=None, pattern="^(human|service|unclassified)?$",
        description=(
            "Classification filter. ``unclassified`` → nobody has said "
            "what this account is yet."
        ),
    ),
    created_after: date | None = Query(
        default=None, description="ISO date (inclusive)",
    ),
    created_before: date | None = Query(
        default=None, description="ISO date (exclusive)",
    ),
    last_login_after: date | None = Query(
        default=None, description="ISO date (inclusive)",
    ),
    last_login_before: date | None = Query(
        default=None, description="ISO date (exclusive)",
    ),
    never_logged_in: bool = Query(
        default=False,
        description="Only IAM users who have never signed in.",
    ),
    stale_days: int | None = Query(
        default=None, ge=1, le=3650,
        description=(
            "Only IAM users whose last_login_at is older than this many "
            "days, OR who have never logged in. Useful for the stale-"
            "account compliance review."
        ),
    ),
    policy: str | None = Query(
        default=None,
        description=(
            "Substring match on attached_policies[*].name (case-"
            "insensitive). Backed by MySQL JSON_SEARCH."
        ),
    ),
    group: str | None = Query(
        default=None,
        description=(
            "Substring match on groups[*] (case-insensitive)."
        ),
    ),
    mine: bool = Query(
        default=False,
        description=(
            "Self-service filter. ``true`` restricts the listing to "
            "IAM users owned by the caller — either explicitly via "
            "``owner_person_id`` or by email-local-part match on "
            "``user_name`` (the IAM username). Used by the employee "
            "``/my/cloud/iam`` view."
        ),
    ),
    revoked: str | None = Query(
        default=None, pattern="^(active|revoked|all)?$",
        description=(
            "Lifecycle filter. ``active`` (default) → only IAM users "
            "currently live cloud-side. ``revoked`` → only IAM users "
            "Helix has cleaned (offboarding revoke / scope-decommission "
            "evacuation). ``all`` → both, with the row carrying "
            "``revoked_at`` to distinguish."
        ),
    ),
    page: int = 1,
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List IAM users across cloud tenants.

    ``mine=true`` is allowed for any logged-in user (employees get a
    self-service view of their own IAM accounts). Otherwise the usual
    ``can_read_cloud`` gate applies.

    Filters: ``integration_id`` (one tenant), ``search`` (user_name /
    arn substring), ``mfa`` ("true" / "false"), ``has_owner``
    ("true"/"false"), ``created_after`` / ``created_before``,
    ``last_login_after`` / ``last_login_before``, ``never_logged_in``,
    ``stale_days`` (no login in N days OR never), ``policy`` /
    ``group`` (JSON substring), ``mine`` (see above).
    """
    if not mine and not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from datetime import datetime as _dt
    from datetime import timedelta as _td

    from sqlalchemy import func as _func
    from sqlalchemy import or_

    from app.domains.cmdb.models import CmdbIamUser

    # Lifecycle filter — default is ``active`` (currently-live cloud
    # IAM users). The IAM users page ``revoked``/``all`` tabs flip this
    # so admins can audit which users Helix has cleaned and when.
    #
    # Note on the ``deleted_at`` interplay: the IAM sync soft-deletes
    # rows that fall off the cloud-side listing (``deleted_at`` stamped
    # on the next sync after ``admin_cleanup`` ran). For the ``active``
    # tab we exclude soft-deleted rows; but the ``revoked`` and ``all``
    # tabs MUST include them, otherwise the cleanup audit trail is
    # invisible the moment the next sync sweeps. Bug fix 2026-04-30:
    # before this, ``deleted_at IS NULL`` was unconditional and the
    # "已被清理" tab always showed 0 once the sweep ran.
    q = select(CmdbIamUser)
    revoked_mode = revoked or "active"
    if revoked_mode == "active":
        q = q.where(
            CmdbIamUser.deleted_at.is_(None),
            CmdbIamUser.revoked_at.is_(None),
        )
    elif revoked_mode == "revoked":
        q = q.where(CmdbIamUser.revoked_at.isnot(None))
    # ``all`` → no lifecycle filter; surfaces every row the admin can
    # legitimately audit, with ``deleted_at`` / ``revoked_at`` carried
    # to the row so the UI can render the lifecycle badge.
    if integration_id:
        q = q.where(CmdbIamUser.integration_id == integration_id)
    elif revoked_mode == "active" and not mine:
        # Hide disabled-tenant IAM rows from the admin "active" tab —
        # the revoke / cleanup buttons all hit live cloud APIs, so
        # showing rows from a disabled tenant invites admins to click
        # revoke against credentials Helix can't actually call. The
        # ``revoked`` and ``all`` tabs are explicit audit views and
        # legitimately include disabled-tenant history.
        #
        # ``mine=True`` is exempt: the /my/cloud/iam self-service view
        # has no revoke buttons (employees can't decommission their
        # own credentials), so the "click-revoke-on-ghost" footgun
        # doesn't apply. Hiding an employee's own IAM accounts because
        # the tenant got disabled would be a worse UX failure — they
        # need to know what credentials Helix has on file even when a
        # tenant goes through admin maintenance.
        active_iam = await _active_cloud_integration_ids(db)
        if not active_iam:
            return paginated_response(data=[], total=0, page=page, page_size=page_size, request=request)
        q = q.where(CmdbIamUser.integration_id.in_(active_iam))
    if search:
        # ``_esc`` neutralizes ``%``/``_``/``\\`` so a malicious search
        # can't force a full-table-scan or pattern-bypass. Same helper
        # is used everywhere else in the cmdb service layer.
        pat = f"%{_esc(search)}%"
        q = q.where(or_(
            CmdbIamUser.user_name.ilike(pat),
            CmdbIamUser.arn.ilike(pat),
        ))
    if mfa == "true":
        q = q.where(CmdbIamUser.mfa_enabled.is_(True))
    elif mfa == "false":
        q = q.where(CmdbIamUser.mfa_enabled.is_(False))
    if has_owner == "true":
        q = q.where(CmdbIamUser.owner_person_id.isnot(None))
    elif has_owner == "false":
        # "Orphan" = no owner AND not a declared robot. A service account
        # will never have an owner; keeping it here means the compliance
        # list can never reach zero, and a list that can never reach zero
        # stops being read.
        q = q.where(
            CmdbIamUser.owner_person_id.is_(None),
            or_(
                CmdbIamUser.account_kind.is_(None),
                CmdbIamUser.account_kind != "service",
            ),
        )
    if account_kind:
        if account_kind == "unclassified":
            q = q.where(CmdbIamUser.account_kind.is_(None))
        else:
            q = q.where(CmdbIamUser.account_kind == account_kind)
    if created_after is not None:
        q = q.where(CmdbIamUser.create_date >= created_after)
    if created_before is not None:
        q = q.where(CmdbIamUser.create_date < created_before)
    if last_login_after is not None:
        q = q.where(CmdbIamUser.last_login_at >= last_login_after)
    if last_login_before is not None:
        q = q.where(CmdbIamUser.last_login_at < last_login_before)
    if never_logged_in:
        q = q.where(CmdbIamUser.last_login_at.is_(None))
    if stale_days is not None:
        cutoff = _dt.now(UTC) - _td(days=stale_days)
        q = q.where(or_(
            CmdbIamUser.last_login_at < cutoff,
            CmdbIamUser.last_login_at.is_(None),
        ))
    if policy:
        # JSON_SEARCH('one', pat, NULL, '$[*].name') returns the path or
        # NULL; matches case-insensitively on LIKE semantics.
        pat = f"%{policy}%"
        q = q.where(_func.json_search(
            CmdbIamUser.attached_policies, "one", pat, None, "$[*].name",
        ).isnot(None))
    if group:
        pat = f"%{group}%"
        q = q.where(_func.json_search(
            CmdbIamUser.groups, "one", pat,
        ).isnot(None))
    if mine:
        # Self-scope: owner FK match ONLY. The previous email-local-part
        # fallback (``user_name == local`` / ``user_name == email``)
        # would surface IAM users from OTHER tenants whose user_name
        # collided with the caller's email-local-part — leaking ARN +
        # attached_policies + groups + MFA across tenants. AWS IAM
        # user_name is free-form, so a service account named "shared"
        # or "admin" or anyone's ex-coworker first-name was visible
        # to anyone whose email started with that string. The IAM
        # reconcile job is the only legitimate way to populate
        # owner_person_id; if it hasn't run, ``mine=true`` returns
        # nothing rather than guessing.
        q = q.where(CmdbIamUser.owner_person_id == str(user.id))
    count_q = select(_func.count()).select_from(q.subquery())
    total = (await db.execute(count_q)).scalar() or 0
    q = (
        q.order_by(CmdbIamUser.integration_id, CmdbIamUser.user_name)
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    result = await db.execute(q)
    rows = list(result.scalars().all())
    # Resolve owner labels (reuse the helper used by CSV export).
    owner_labels = await _resolve_owner_labels(
        db, [r.owner_person_id for r in rows if r.owner_person_id],
    )
    owner_meta = await _resolve_owner_meta(
        db, [r.owner_person_id for r in rows if r.owner_person_id],
    )
    # Resolve provider + integration_name for display. Without these
    # fields the employee ``/my/cloud/iam`` view can't tell a Tencent
    # row apart from an AWS one.
    integ_ids = {str(r.integration_id) for r in rows}
    integ_map: dict[str, Integration] = {}
    if integ_ids:
        integ_rows = (
            await db.execute(
                select(Integration).where(
                    Integration.id.in_([uuid.UUID(i) for i in integ_ids]),
                )
            )
        ).scalars().all()
        integ_map = {str(i.id): i for i in integ_rows}
    data = []
    for r in rows:
        item = IamUserResponse.model_validate(r).model_dump()
        item["console_login"] = _console_login_of(r)
        if r.owner_person_id:
            item["owner_label"] = owner_labels.get(r.owner_person_id)
            m = owner_meta.get(r.owner_person_id)
            if m:
                item["owner_employee_no"] = m["employee_no"]
                item["owner_employment_status"] = m["employment_status"]
        integ = integ_map.get(str(r.integration_id))
        if integ is not None:
            item["provider"] = integ.provider
            item["integration_name"] = integ.name
        data.append(item)
    return paginated_response(
        data=data, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/iam-users/{iam_user_id}")
async def get_iam_user(
    iam_user_id: uuid.UUID,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Single IAM user by id.

    Mirrors the ``load-balancers/{id}`` by-id pattern + the iam list's
    ``mine`` ownership rule: cloud readers (admin/auditor) see any row;
    everyone else only their OWN (``owner_person_id == self``). Added so
    the mobile ``/my/cloud/iam`` detail deep-link resolves by id instead
    of paging the list and ``.find()``-ing (which silently 404s for a
    holder of >100 IAM accounts).
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbIamUser
    row = (await db.execute(
        select(CmdbIamUser).where(CmdbIamUser.id == iam_user_id),
    )).scalar_one_or_none()
    if row is None:
        raise NotFoundError("IamUser", str(iam_user_id))
    # Per-row scope: non-readers may only see their own (owner FK match,
    # same as the list's ``mine`` filter — no email-local-part guessing).
    if not can_read_cloud(user.role_names):
        if row.owner_person_id != str(user.id):
            raise NotFoundError("IamUser", str(iam_user_id))
    item = IamUserResponse.model_validate(row).model_dump()
    item["console_login"] = _console_login_of(row)
    if row.owner_person_id:
        labels = await _resolve_owner_labels(db, [row.owner_person_id])
        meta = await _resolve_owner_meta(db, [row.owner_person_id])
        item["owner_label"] = labels.get(row.owner_person_id)
        m = meta.get(row.owner_person_id)
        if m:
            item["owner_employee_no"] = m["employee_no"]
            item["owner_employment_status"] = m["employment_status"]
    integ = (await db.execute(
        select(Integration).where(Integration.id == row.integration_id),
    )).scalar_one_or_none()
    if integ is not None:
        item["provider"] = integ.provider
        item["integration_name"] = integ.name
    return success_response(data=item, request=request)


@router.patch("/iam-users/{iam_user_id}/owner")
async def update_iam_user_owner(
    request: Request,
    iam_user_id: str,
    body: IamUserOwnerUpdate,
    user: AuthUser = Depends(get_current_user),
    # Pre-launch P0 fix (Round 45 follow-up, CR-209): IAM ownership
    # re-bind is a high-trust principal mutation — a stolen PAT
    # must not be sufficient. ``cleanup_iam_user`` already had this
    # gate; ``update_iam_user_owner`` was inconsistent. Browser
    # session required; MCP path (if added later) still works via
    # the service layer. Declared AFTER ``get_current_user`` so
    # ``auth_via`` is set before the gate reads it.
    _gate: None = Depends(require_browser_session),
    db: AsyncSession = Depends(get_db),
):
    """Re-bind a CMDB IAM user to a different Helix person (or unbind).

    The IAM-sync job (e.g. ``cmdb.iam_sync``) populates ``owner_person_id``
    automatically by best-effort heuristics on the IAM user's name +
    email. When the heuristic gets it wrong (a service account that
    happens to match someone's email-local-part, or an IAM user that was
    handed off without a rename), an admin needs to override it manually.
    Surfaces this on the ``/admin/cloud/iam-users`` page.

    Note: this only changes Helix-side ownership metadata. It does NOT
    rotate keys, change cloud-side IAM, or notify the previous owner —
    use the dedicated revoke / grant flows for that.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError(
            "Only cloud-admin / global_admin can re-bind IAM user ownership.",
        )
    from app.domains.cmdb.models import CmdbIamUser
    from app.identity.person_models import IdentityPerson

    iam_row = await db.get(CmdbIamUser, uuid.UUID(iam_user_id))
    if iam_row is None or iam_row.deleted_at is not None:
        raise NotFoundError("CmdbIamUser", iam_user_id)

    new_owner: IdentityPerson | None = None
    if body.owner_person_id is not None:
        new_owner = await db.get(IdentityPerson, body.owner_person_id)
        if new_owner is None:
            raise NotFoundError(
                "IdentityPerson", str(body.owner_person_id),
            )
        # Refuse re-binding to a leaver — same defence the IAM-cleanup
        # transfer path enforces. Without this, an admin can transfer
        # IAM ownership to an already-departed employee and silently
        # defeat offboarding (the leaver still owns nothing, but Helix
        # records them as the owner of an active IAM principal).
        # Pre-launch P1 fix (Round 45, CR-203): refuse leaver_pending
        # too. assign_scope_owner already rejects both states; this
        # path was inconsistent — admin could seat a soon-to-leave
        # employee as IAM principal owner, then T+0 sweep faces an
        # inconsistent owner FK.
        if (
            new_owner.deleted_at is not None
            or new_owner.employment_status in ("left", "leaver_pending")
        ):
            raise AppException(
                code="IAM_OWNER_TARGET_INACTIVE",
                message=(
                    f"Cannot bind IAM user to {new_owner.work_email} — "
                    f"person is {new_owner.employment_status or 'deleted'} "
                    "in HR. Pick a current employee or unset the owner."
                ),
                status_code=409,
            )

    previous_owner_id = iam_row.owner_person_id
    iam_row.owner_person_id = (
        str(new_owner.id) if new_owner is not None else None
    )

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=user.id,
        actor_type="user",
        action="cmdb.iam_user.owner.update",
        resource_type="cmdb_iam_user",
        resource_id=str(iam_row.id),
        status="success",
        detail={
            "iam_user_name": iam_row.user_name,
            "integration_id": str(iam_row.integration_id),
            "previous_owner_person_id": previous_owner_id,
            "new_owner_person_id": (
                str(new_owner.id) if new_owner else None
            ),
            "reason": body.reason,
        },
        ip_address=get_real_ip(request),
    )
    await db.commit()
    await db.refresh(iam_row)
    return success_response(
        data=await _enrich_iam_user_response(db, iam_row),
        request=request,
    )


def _console_login_of(iam_row) -> bool | None:
    """Provider console-login switch, lifted out of ``raw_data``.

    Tencent CAM returns ``ConsoleLogin`` 0/1; ``0`` means the sub-user
    cannot sign in at all, which is the authoritative "this is a robot"
    signal (``tiup-server``, ``operator-v1-cos-upload``). Returns None
    when the provider does not report it, so the UI can tell "no console
    access" from "we do not know".
    """
    raw = iam_row.raw_data or {}
    value = raw.get("ConsoleLogin")
    if value is None:
        return None
    return bool(value)


async def _enrich_iam_user_response(
    db: AsyncSession, iam_row,
) -> dict:
    """Render one ``CmdbIamUser`` row with the same enrichment shape the
    list endpoint returns: provider, integration_name, owner_label.

    The frontend re-uses this single-row response to in-place replace the
    matching row in its in-memory list (so the user sees the new state
    without a refetch). If we returned the raw model the row would lose
    its provider tag + owner display until the next refetch — visible
    flicker on every cleanup / owner-change action.
    """
    payload = IamUserResponse.model_validate(iam_row).model_dump()
    payload["console_login"] = _console_login_of(iam_row)
    if iam_row.owner_person_id:
        labels = await _resolve_owner_labels(db, [iam_row.owner_person_id])
        payload["owner_label"] = labels.get(iam_row.owner_person_id)
        owner_meta = await _resolve_owner_meta(db, [iam_row.owner_person_id])
        m = owner_meta.get(str(iam_row.owner_person_id))
        if m:
            payload["owner_employee_no"] = m["employee_no"]
            payload["owner_employment_status"] = m["employment_status"]
    integ = await db.get(
        Integration, uuid.UUID(str(iam_row.integration_id)),
    )
    if integ is not None:
        payload["provider"] = integ.provider
        payload["integration_name"] = integ.name
    return payload


async def _audit_iam_cleanup_failure(
    *,
    trace_id: str,
    actor_user_id: uuid.UUID,
    iam_user_id: str,
    detail: dict,
) -> None:
    """Persist a ``cmdb.iam_user.cleanup.disown/failed`` audit row in a
    fresh session so the forensic trail survives the rollback caused by
    the ``AppException`` we are about to raise.

    Mirrors :func:`_audit_decommission_failure` in lifecycle_service —
    same justification: a destructive admin action that fails on the
    cloud side must leave a record of *who tried what and why it failed*
    even though the request transaction will be rolled back.
    """
    from app.core.database import async_session_factory

    try:
        async with async_session_factory() as audit_session:
            audit = AuditService(audit_session)
            await audit.log(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=actor_user_id,
                action="cmdb.iam_user.cleanup.disown",
                status="failed",
                resource_type="cmdb_iam_user",
                resource_id=iam_user_id,
                detail=detail,
            )
            await audit_session.commit()
    except Exception:  # noqa: BLE001 — best-effort logging

        # exc_info=False — see ``_set_dns_record_status_direct``:
        # cloud-provider SDK exception args sometimes carry signed
        # request URLs (with AccessKeyId + signature) that we don't
        # want leaking into ELK.
        get_safe_logger(__name__).warning(
            "Out-of-tx audit write for IAM cleanup failure on %s failed",
            iam_user_id,
        )


@router.post("/iam-users/{iam_user_id}/cleanup")
async def cleanup_iam_user(
    request: Request,
    iam_user_id: str,
    body: IamUserCleanupRequest,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_browser_session),
):
    """Clean up an IAM user — irreversible cloud-side delete OR transfer.

    Two flows:

    * ``transfer_to_person_id`` set → **transfer** mode. Re-bind the
      Helix ownership FK to the named active person; the IAM user keeps
      working on the cloud side. Helix-side bookkeeping only.
    * ``transfer_to_person_id`` null → **disown** mode. Run the cloud
      driver's full IAM revoke chain (login profile, access keys,
      attached policies, group memberships, MFA, signing certs, SSH
      keys, then ``DeleteUser``) and only on driver ``ok=True`` clear
      Helix ownership and stamp ``revoked_at`` + ``revoke_reason``.

    Disown is **irreversible**: AWS / Aliyun / Tencent / UCloud all
    invalidate every signing credential the moment ``DeleteUser`` is
    called. The frontend wraps the trigger in a Popconfirm with a red
    warning so the human sees the warning a second time.

    Failure semantics:

    * ``revoke_account`` returns ``ok=False`` → 502
      ``IAM_CLEANUP_PROVIDER_ERROR`` with the provider's verbatim
      error_message; the row is left untouched in Helix so the admin
      can fix the precondition (missing IAM permissions, etc.) and
      retry. The failed attempt is audit-logged via a fresh session so
      the forensic trail survives the rollback.
    * Driver does not implement ``revoke_account`` → 501
      ``IAM_CLEANUP_NOT_SUPPORTED``. Currently AWS / Aliyun / Tencent /
      UCloud all implement it (those are the only providers that
      populate ``cmdb_iam_users``); a fresh provider that surfaces an
      IAM list but not a delete chain would surface this 501 rather
      than silently no-op.

    Idempotency: AWS ``NoSuchEntity`` short-circuits inside the driver
    with ``ok=True`` (note: ``raw_response.note = 'user not found —
    already deleted'``). Calling cleanup twice on the same row is safe
    — the second call sees ``revoked_at`` already set and the driver
    treats the missing user as success.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError(
            "You do not have permission to manage cloud IAM users.",
        )
    from app.domains.cmdb.lifecycle_service import _load_cloud_driver
    from app.domains.cmdb.models import CmdbIamUser
    from app.identity.person_models import IdentityPerson
    from app.integrations.driver_base import (
        CloudAccountRevokeContext,
        ProvisioningResult,
    )

    iam_row = await db.get(CmdbIamUser, uuid.UUID(iam_user_id))
    if iam_row is None or iam_row.deleted_at is not None:
        raise NotFoundError("CmdbIamUser", iam_user_id)

    new_owner: IdentityPerson | None = None
    if body.transfer_to_person_id is not None:
        new_owner = await db.get(IdentityPerson, body.transfer_to_person_id)
        if new_owner is None:
            raise NotFoundError(
                "IdentityPerson", str(body.transfer_to_person_id),
            )
        # Pre-launch P1 fix (Round 45, CR-204): reject leaver_pending
        # AND left. The whole point of this endpoint is to disown FROM
        # a leaver; re-binding INTO another leaver / leaver_pending
        # just defers the same problem and breaks parity with
        # assign_scope_owner.
        if new_owner.employment_status in ("left", "leaver_pending"):
            raise ForbiddenError(
                f"Refusing to transfer IAM user to a "
                f"{new_owner.employment_status} person — pick an "
                "active employee.",
            )

    previous_owner_id = iam_row.owner_person_id

    raw = dict(iam_row.raw_data or {})
    helix_meta = dict(raw.get("_helix") or {})
    revoke_steps: list[str] | None = None

    if new_owner is None:
        # ── Disown path: real cloud-side delete via the driver's IAM
        # revoke chain. This is irreversible (DeleteUser invalidates every
        # signing credential), so reserve it for global_admin — matching
        # ``can_manage_cloud_scope_lifecycle`` (irreversible cloud lifecycle
        # is global_admin-only, per rbac.py + CLAUDE.md) and this endpoint's
        # own historical error message. ``cloud_admin`` keeps the reversible
        # transfer path above; only the destructive disown is gated up.
        if not can_manage_cloud_scope_lifecycle(user.role_names):
            raise ForbiddenError(
                "Only global_admin can disown (irreversibly delete) a "
                "cloud IAM user. cloud_admin may transfer ownership "
                "instead.",
            )
        # Refuse to re-run if the row is already revoked (idempotent
        # short-circuit; protects against accidental double-clicks racing
        # past the in-memory check).
        if iam_row.revoked_at is not None:
            return success_response(
                data=await _enrich_iam_user_response(db, iam_row),
                request=request,
            )

        integration = await db.get(
            Integration, uuid.UUID(str(iam_row.integration_id)),
        )
        if integration is None or integration.kind != "cloud":
            raise NotFoundError(
                "CloudIntegration", str(iam_row.integration_id),
            )

        # Production kill-switch. Mirror ``_execute_server_lifecycle`` —
        # the global prod-write lock that blocks server destroy must also
        # block an irreversible IAM ``DeleteUser`` during an incident
        # lockdown. An IAM user inherits its environment from the cloud
        # account (shared ``integration_id`` on CmdbCloudScope); treat it
        # as prod if the integration has ANY active prod scope (fail-safe).
        from app.domains.cmdb.models import (
            PROD_ENVIRONMENTS,
            CmdbCloudScope,
        )
        from app.domains.cmdb.prod_write_lock import (
            LOCKED_MESSAGE,
            prod_cloud_writes_enabled,
        )
        has_prod_scope = (await db.execute(
            select(CmdbCloudScope.id).where(
                CmdbCloudScope.integration_id == str(iam_row.integration_id),
                CmdbCloudScope.environment.in_(tuple(PROD_ENVIRONMENTS)),
                CmdbCloudScope.deleted_at.is_(None),
            ).limit(1)
        )).first() is not None
        if has_prod_scope and not await prod_cloud_writes_enabled(db):
            raise ProdCloudLockedError(LOCKED_MESSAGE)

        driver = await _load_cloud_driver(db, integration)
        ctx = CloudAccountRevokeContext(
            provider=integration.provider,
            account_id=iam_row.user_name,
        )

        try:
            result: ProvisioningResult = await driver.revoke_account(ctx)
        except NotImplementedError as exc:
            await _audit_iam_cleanup_failure(
                trace_id=request.state.trace_id,
                actor_user_id=user.id,
                iam_user_id=str(iam_row.id),
                detail={
                    "iam_user_name": iam_row.user_name,
                    "integration_id": str(iam_row.integration_id),
                    "provider": integration.provider,
                    "reason": body.reason,
                    "error_code": "not_supported",
                    "error_message": str(exc),
                },
            )
            raise AppException(
                code="IAM_CLEANUP_NOT_SUPPORTED",
                message=(
                    f"Cloud provider {integration.provider!r} does not "
                    "support IAM-level revocation; clean up the user via "
                    "the provider console."
                ),
                status_code=501,
            ) from exc
        finally:
            await driver.close()

        if not result.ok:
            await _audit_iam_cleanup_failure(
                trace_id=request.state.trace_id,
                actor_user_id=user.id,
                iam_user_id=str(iam_row.id),
                detail={
                    "iam_user_name": iam_row.user_name,
                    "integration_id": str(iam_row.integration_id),
                    "provider": integration.provider,
                    "reason": body.reason,
                    "error_message": (
                        result.error_message or "driver returned ok=False"
                    ),
                    "raw_response": result.raw_response,
                },
            )
            raise AppException(
                code="IAM_CLEANUP_PROVIDER_ERROR",
                message=(
                    result.error_message
                    or "Cloud provider rejected the IAM revocation."
                ),
                status_code=502,
            )

        steps = (result.raw_response or {}).get("steps")
        if isinstance(steps, list):
            revoke_steps = [str(s) for s in steps]
        iam_row.owner_person_id = None
        iam_row.revoked_at = datetime.now(UTC)
        # 64-char column → keep the reason tag short + machine-readable.
        # The full free-text reason is preserved in the audit log and in
        # ``_helix.cleanup_reason`` below.
        iam_row.revoke_reason = "admin_cleanup"
        helix_meta.pop("cleanup_pending", None)
        helix_meta["cleanup_reason"] = body.reason
        helix_meta["cleanup_marked_at"] = datetime.now(UTC).isoformat()
        helix_meta["cleanup_marked_by_person_id"] = str(user.id)
        if revoke_steps is not None:
            helix_meta["cleanup_revoke_steps"] = revoke_steps

        # ── Cascade revoke of grants attributed to this IAM user ──
        #
        # Without this block, ``cmdb_cloud_access_grants`` rows for the
        # IAM user we just deleted cloud-side would keep showing
        # ``status='active'`` until the next sync drift sweep. Two
        # specific failure modes that drove this cascade:
        #
        # 1. ``source='helix'`` grants are NEVER touched by the drift
        #    sweep (Helix is system-of-record for those — see
        #    ``sync_grants_for_integration`` lifecycle_service.py:2977).
        #    Result: ghost helix grants forever for a deleted IAM user.
        # 2. The drift sweep's 50%-safety-floor refuses to fire when a
        #    big batch of cleanups would close >50% of discovered
        #    grants (defends against driver pagination truncation).
        #    Mass IAM cleanup hits this floor and the discovered grants
        #    also get stranded.
        #
        # Bug report 2026-05-05: AWS SG (Singapore) "成员 / 生效 grant
        # 37" persisted after every IAM user was disowned + a manual
        # 立即同步 was triggered.
        #
        # Two-pass cascade:
        #   (a) discovered grants — match by external_ref prefix
        #       ``iam-user:{user_name}::``. ``_extract_iam_user_grants``
        #       in lifecycle_service.py stamps this on every row it
        #       upserts, so the prefix is the canonical FK back to the
        #       IAM user.
        #   (b) helix-issued grants for the same person on the scopes
        #       those discovered rows live on. The IAM user was the
        #       auth handle for everything that person had on those
        #       scopes — deleting the IAM user invalidates those
        #       helix-mediated permissions too. Without (b), helix
        #       grants would never close.
        #
        # All cascades stamp ``revoke_reason='iam_user_disowned'`` so
        # forensics can distinguish from natural ``user_released`` /
        # ``inactive`` / ``drift_removed`` revocations.
        from sqlalchemy import or_

        from app.domains.cmdb.models import (
            GRANT_REVOKE_REASON_IAM_DISOWNED,
            GRANT_STATUS_ACTIVE,
            GRANT_STATUS_REVOKED,
            CmdbCloudAccessGrant,
            CmdbCloudScope,
        )
        # Critical: scope the cascade to scopes belonging to THIS
        # integration. Two AWS integrations (e.g. AWS prod + AWS
        # staging) can both have an IAM user named ``li.da`` — without
        # the integration filter, disowning one tenant's IAM user would
        # cascade-revoke the other tenant's grants. ``CmdbIamUser`` is
        # tied to integration_id (not scope_id), so the safe cone is
        # "active scopes that belong to this integration".
        integration_scope_ids: set[str] = {
            str(sid) for sid in (await db.execute(
                select(CmdbCloudScope.id).where(
                    CmdbCloudScope.integration_id == str(iam_row.integration_id),
                    CmdbCloudScope.deleted_at.is_(None),
                )
            )).scalars().all()
        }
        external_ref_prefix = f"iam-user:{iam_row.user_name}::"
        # Escape SQL LIKE wildcards in ``user_name`` so a name like
        # ``svc_admin`` (Aliyun RAM allows ``_``; AWS IAM allows ``_``)
        # doesn't accidentally widen the prefix into a wildcard. Used
        # by both LIKE branches below — defining it once keeps them in
        # sync (the two branches diverging is a real footgun, see PR
        # review M1 / 2026-05-05).
        escaped_prefix = (
            external_ref_prefix.replace("\\", "\\\\")
            .replace("%", "\\%").replace("_", "\\_")
        ) + "%"
        # Pre-launch P1 fix (Round 45, CR-206 + follow-up CR-210):
        # include PENDING grants too in BOTH the discovered-grants
        # branch and the helix-grants branch below. Previously the
        # cascade only revoked ACTIVE grants, leaving pending rows
        # pointing at a now-deleted IAM principal — approving them
        # later would 502. Hoist the import so the helix-grants
        # branch can reference it even if the discovered branch
        # skipped (defensive — affected_scope_ids being non-empty
        # implies discovered ran, but explicit > implicit).
        from app.domains.cmdb.models import (  # noqa: PLC0415
            GRANT_STATUS_PENDING,
        )

        discovered_grants: list[CmdbCloudAccessGrant] = []
        if integration_scope_ids:
            discovered_grants = list((await db.execute(
                # ``with_for_update`` so a concurrent grant_access /
                # renew that lands between this SELECT and the
                # transaction commit gets serialised — without the
                # lock, the cascade snapshot could miss a grant that
                # was issued mid-cleanup and the new grant would
                # survive pointing at a now-deleted IAM user.
                select(CmdbCloudAccessGrant).where(
                    CmdbCloudAccessGrant.status.in_((
                        GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING,
                    )),
                    # Constrain to this integration's scopes — without
                    # this, a same-named IAM user in another tenant
                    # would have its grants caught too.
                    CmdbCloudAccessGrant.scope_id.in_(integration_scope_ids),
                    CmdbCloudAccessGrant.external_ref.like(
                        escaped_prefix, escape="\\",
                    ),
                ).with_for_update()
            )).scalars().all())
        affected_scope_ids = {str(g.scope_id) for g in discovered_grants}
        helix_grants: list[CmdbCloudAccessGrant] = []
        if affected_scope_ids and previous_owner_id:
            # Pre-launch P1 fix (Round 45 follow-up, CR-210): include
            # PENDING here too — the discovered-grants branch above
            # already does. A pending helix-source grant pointing at
            # the now-deleted IAM principal would 502 on later
            # approval. Same family fix as CR-206, helix branch.
            helix_grants = list((await db.execute(
                select(CmdbCloudAccessGrant).where(
                    CmdbCloudAccessGrant.status.in_((
                        GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING,
                    )),
                    CmdbCloudAccessGrant.person_id == previous_owner_id,
                    # Same integration constraint as discovered branch
                    # — affected_scope_ids is already a subset of
                    # integration_scope_ids by construction, but pin
                    # it explicitly so a future refactor that widens
                    # the discovered query can't accidentally widen
                    # this one too.
                    CmdbCloudAccessGrant.scope_id.in_(affected_scope_ids),
                    CmdbCloudAccessGrant.source == "helix",
                    # Skip rows already caught by the discovered query
                    # (defensive — external_ref is unique-by-construction
                    # so this should be a no-op, but a future migration
                    # that decoupled source from external_ref schema
                    # would otherwise double-process). Use the SAME
                    # escaped pattern as the discovered branch so a
                    # user_name with ``_`` or ``%`` doesn't behave
                    # asymmetrically across the two LIKEs.
                    or_(
                        CmdbCloudAccessGrant.external_ref.is_(None),
                        ~CmdbCloudAccessGrant.external_ref.like(
                            escaped_prefix, escape="\\",
                        ),
                    ),
                ).with_for_update()
            )).scalars().all())
        cascade_now = datetime.now(UTC)
        cascaded_grant_ids: list[str] = []
        for g in [*discovered_grants, *helix_grants]:
            g.status = GRANT_STATUS_REVOKED
            g.revoked_at = cascade_now
            g.revoke_reason = GRANT_REVOKE_REASON_IAM_DISOWNED
            # Clear any stale inactivity-stage marker so the inactivity
            # sweep can't pick the row back up via leftover state.
            g.last_notified_stage = None
            cascaded_grant_ids.append(str(g.id))
        helix_meta["cascade_revoked_grant_ids"] = cascaded_grant_ids
        helix_meta["cascade_revoked_grant_count"] = len(cascaded_grant_ids)
    else:
        # ── Transfer path: Helix-side rebind only. Clears any prior
        # pending-cleanup mark so the row no longer shows the warning.
        iam_row.owner_person_id = str(new_owner.id)
        helix_meta.pop("cleanup_pending", None)
        helix_meta["last_transfer_reason"] = body.reason
        helix_meta["last_transfer_to_person_id"] = str(new_owner.id)
        helix_meta["last_transfer_at"] = datetime.now(UTC).isoformat()

    raw["_helix"] = helix_meta
    iam_row.raw_data = raw

    audit = AuditService(db)
    # ``cascaded_grant_ids`` is set inside the disown branch only;
    # transfer keeps it None so the audit row remains terse.
    cascaded_for_audit: list[str] | None = (
        helix_meta.get("cascade_revoked_grant_ids")
        if new_owner is None else None
    )
    detail: dict[str, Any] = {
        "iam_user_name": iam_row.user_name,
        "integration_id": str(iam_row.integration_id),
        "previous_owner_person_id": previous_owner_id,
        "new_owner_person_id": (
            str(new_owner.id) if new_owner else None
        ),
        "reason": body.reason,
    }
    if revoke_steps is not None:
        detail["revoke_steps"] = revoke_steps
    if cascaded_for_audit is not None:
        detail["cascade_revoked_grant_count"] = len(cascaded_for_audit)
        # Truncate the id list at 200 — the count is the primary
        # signal; the full list lives in ``iam_row.raw_data._helix``
        # for forensic queries.
        detail["cascade_revoked_grant_ids"] = cascaded_for_audit[:200]
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=user.id,
        actor_type="user",
        action=(
            "cmdb.iam_user.cleanup.transfer"
            if new_owner is not None
            else "cmdb.iam_user.cleanup.disown"
        ),
        resource_type="cmdb_iam_user",
        resource_id=str(iam_row.id),
        status="success",
        detail=detail,
        ip_address=get_real_ip(request),
    )
    await db.commit()
    await db.refresh(iam_row)
    return success_response(
        data=await _enrich_iam_user_response(db, iam_row),
        request=request,
    )


# ---------------------------------------------------------------------------
# Billing
# ---------------------------------------------------------------------------


@router.get("/billing/monthly")
async def list_monthly_billing(
    request: Request,
    integration_id: str | None = None,
    months: int = Query(default=6, ge=1, le=24),
    measure: str | None = Query(
        default=None,
        description=(
            "成本口径:``list``(默认,业务折前成本/研发口径,按月聚合自 "
            "cmdb_cloud_billing_daily)或 ``invoice``(实付,读 "
            "cmdb_billing_monthly,仅 finance / cloud_admin / "
            "global_admin)。"
        ),
    ),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List recent monthly billing rows (one row per integration × month).

    Readable by cloud-admin AND finance (Finance role has read-only
    scope on every billing-adjacent screen per the HR+Finance roles
    ADR, 2026-04-21).

    口径路由（2026-07 保密红线补口）：``cmdb_billing_monthly`` 是
    **实付**（发票）口径的表，没有 list 列。账单页的日粒度总览已按
    研发口径路由，同页的 12 个月趋势若继续裸读实付表，两口径同框、
    月粒度折扣率即可导出 —— 所以默认（list）改为从
    ``cmdb_cloud_billing_daily`` 用 ``daily_engineering_amount()``
    （COALESCE(list, invoice) 过渡语义）按月聚合；``invoice`` 保持
    原表、仅保密圈可请求。日表行数 = integration × scope × 日（无
    宽 GROUP BY 分块问题——小表）。``cost_by_service`` 两个口径都恒
    ``None`` —— 理由见下方 payload 处的注释（list 的日表拆分是实付
    口径;invoice 的月表拆分币种在读侧不可判定）。
    """
    if not (can_read_cloud(user.role_names) or can_view_finance(user.role_names)):
        raise ForbiddenError()
    # 口径闸:默认 list;invoice 仅保密圈,无权限 → 403(绝不静默降级)。
    from app.domains.cmdb.cost_view import (
        COST_MEASURE_INVOICE,
        resolve_measure,
    )

    try:
        resolved_measure = resolve_measure(measure, user.role_names)
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc
    if resolved_measure is None:
        raise ForbiddenError()
    from datetime import date as _date_t
    from types import SimpleNamespace

    from app.domains.cmdb.cost_router import _waterfall_start_month
    from app.domains.cmdb.models import (
        CmdbBillingMonthly,
        CmdbCloudBillingDaily,
    )

    # Switch from ``LIMIT months * 20`` to a date-cutoff filter. The
    # old shape silently truncated the most-recent-month rows whenever
    # the active-integration count exceeded ~20 (24 tenants × 6 months
    # = 120, but ``LIMIT 120`` after ``ORDER BY month DESC`` cuts at
    # the 5th month — silently losing the 6th month for every tenant).
    today = _date_t.today()
    # ``months`` 个自然月**含当前月**。此前的 ``for _ in range(months)``
    # 从当月月初再退 months 个月,窗口实际是 months+1 个不同的月(FE
    # 的 12 月趋势吐 13 根柱、``le=24`` 的上限执行成 25 个月)。共用
    # 财务瀑布的月算术,不再各写一套。
    anchor = _waterfall_start_month(today, months)

    if resolved_measure == COST_MEASURE_INVOICE:
        q = (
            select(CmdbBillingMonthly)
            .where(CmdbBillingMonthly.month >= anchor)
            .order_by(CmdbBillingMonthly.month.desc())
        )
        if integration_id:
            q = q.where(CmdbBillingMonthly.integration_id == integration_id)
        # Hard-cap to defend against pathological months values; even so,
        # the date predicate already does the heavy lifting.
        q = q.limit(10000)
        result = await db.execute(q)
        rows: list = list(result.scalars().all())
    else:
        # list 口径:按 (integration, 月, currency) 聚合日表。currency
        # 进分组键 —— 日表的 CN 云行是 CNY raw(与 CmdbBillingMonthly
        # 的 legacy 行同一契约:读侧按行 currency 换算),混币合计会把
        # CNY 当 USD 加总。fetched_at 取组内最新,id 用确定性 uuid5
        # (调用方把它当 rowKey,跨请求须稳定)。
        from sqlalchemy import func

        from app.domains.cmdb.cost_view import daily_engineering_amount

        month_expr = func.date_format(
            CmdbCloudBillingDaily.date, "%Y-%m-01",
        ).label("month_str")
        agg_q = (
            select(
                CmdbCloudBillingDaily.integration_id,
                month_expr,
                CmdbCloudBillingDaily.currency,
                func.coalesce(
                    func.sum(daily_engineering_amount()), 0,
                ).label("total"),
                func.max(CmdbCloudBillingDaily.fetched_at).label(
                    "fetched_at",
                ),
            )
            .where(CmdbCloudBillingDaily.date >= anchor)
            .group_by(
                CmdbCloudBillingDaily.integration_id,
                month_expr,
                CmdbCloudBillingDaily.currency,
            )
            .order_by(month_expr.desc())
            .limit(10000)
        )
        if integration_id:
            agg_q = agg_q.where(
                CmdbCloudBillingDaily.integration_id == integration_id,
            )
        rows = [
            SimpleNamespace(
                # rowKey 不含 currency —— 多币种组在下面按 (integration,
                # 月) 折算合并成一行,id 必须与合并后的粒度一致。
                id=uuid.uuid5(
                    uuid.NAMESPACE_URL,
                    f"helix:billing-monthly:list:{r.integration_id}:"
                    f"{r.month_str}",
                ),
                integration_id=r.integration_id,
                month=date.fromisoformat(str(r.month_str)),
                total_cost_usd=float(r.total or 0),
                currency=r.currency,
                fetched_at=r.fetched_at,
            )
            for r in (await db.execute(agg_q)).all()
        ]

    # Enrich with provider + integration_name so the redesigned billing
    # page's monthly stacked-bar can group by cloud without falling back
    # to "unknown" when the period filter narrows ``by_account`` away
    # from older months. Single batched fetch keeps it O(integrations)
    # rather than O(rows).
    integration_ids = {str(r.integration_id) for r in rows}
    integrations: dict[str, Integration] = {}
    if integration_ids:
        for integ in (
            await db.execute(
                select(Integration).where(
                    Integration.id.in_([uuid.UUID(i) for i in integration_ids]),
                )
            )
        ).scalars().all():
            integrations[str(integ.id)] = integ

    # ── 读侧 CNY→USD 折算(两个口径共用) ──────────────────────────
    # ``total_cost_usd`` 列名骗人:CN 云(国内阿里云 / 腾讯 / UCloud)按
    # 契约把**原生 CNY**写进该列并标 ``currency='CNY'``(见
    # lifecycle_service.py 的 provider 汇总与 aliyun_cloud.py 驱动)。
    # 日表没有写侧 rebase;``cmdb_billing_monthly`` 也只有 Round-27
    # 之后的行才在写侧 rebase 成 USD,更早的历史行同样是 CNY。裸加总
    # = 把 ¥ 当 $,~7× 虚高,并且与同一页面上方的 ``/billing/overview``
    # (逐行折算)对不上 —— 同一窗口两个数,cloud_admin 无法判断哪个
    # 是真的。统一走 ``CloudCostUsdConverter``:同一解析顺序
    # (integration override → 平台周更汇率)、同一 2dp 取整。
    #
    # 折不出汇率的币种**绝不**静默按 USD 计入,也不伪造 0:该币种组
    # 的原生额累进 ``meta.unconverted_by_currency``,并且**整个
    # (integration, 月) 行不发** —— 只剔掉折不出的那一组、把剩下的组
    # 当完整合计发出去,等于把一个偏低的数标成「本月全部」,而披露只在
    # 请求级 meta 里(读者无法知道是哪一行缺了多少)。缺行是诚实的;
    # $0 行 / 偏低行都会被读成「这个月这家云就花了这么多」。
    from decimal import ROUND_HALF_UP as _ROUND_HALF_UP
    from decimal import Decimal as _Dec

    from app.core.fx_rate import CloudCostUsdConverter

    cost_conv = CloudCostUsdConverter(db)
    # (integration_id, 月) → 折算后 USD 累加器。invoice 表在
    # (integration, 月) 上唯一,所以那条路径每个 key 只有一行;list
    # 路径的多币种组在这里合并。
    folded: dict[tuple[str, date], SimpleNamespace] = {}
    for r in rows:
        integ = integrations.get(str(r.integration_id))
        if integ is None:
            # Orphaned billing row — the integration was hard-deleted but
            # historical billing kept. Skip rather than emit a row with
            # phantom provider info; the admin can find these via the
            # raw table. Soft-deleted integrations still surface here
            # because the lookup is ``Integration.id == ...`` without a
            # deleted_at filter.
            continue
        row_currency = (r.currency or "USD").upper()
        # 折算是否落地:用 converter 自己的返回信号。此前靠
        # ``unconverted_by_currency`` 有没有增长来判定 —— 累加 0 不改变
        # 字典值,所以「金额恰好 0 的不可折算行」会被误判成折算成功,
        # 发出一条 ``total_cost_usd: 0.0 / currency: USD`` 的假行。
        usd, landed = await cost_conv.to_usd_checked(
            r.total_cost_usd, row_currency, str(r.integration_id),
        )
        key = (str(r.integration_id), r.month)
        acc = folded.get(key)
        if acc is None:
            acc = folded[key] = SimpleNamespace(
                id=r.id,
                integration_id=str(r.integration_id),
                integration_name=integ.name,
                provider=integ.provider,
                month=r.month,
                total=_Dec("0"),
                fetched_at=r.fetched_at,
                any_converted=False,
                any_unconverted=False,
            )
        if r.fetched_at and (
            acc.fetched_at is None or r.fetched_at > acc.fetched_at
        ):
            acc.fetched_at = r.fetched_at
        if not landed:
            acc.any_unconverted = True
            continue
        acc.any_converted = True
        acc.total += _Dec(str(usd))

    data = []
    for acc in sorted(
        folded.values(), key=lambda a: (a.month, a.integration_name),
        reverse=True,
    ):
        # 一个组都没折成 → 缺行;部分组没折成 → 同样缺行(见上方注释:
        # 偏低的合计标成完整,比缺行更容易误导)。
        if not acc.any_converted or acc.any_unconverted:
            continue
        payload = BillingMonthlyResponse(
            id=acc.id,
            integration_id=acc.integration_id,
            integration_name=acc.integration_name,
            provider=acc.provider,
            month=acc.month,
            # 2dp:与写侧 rebase(handlers.py)和 ``/billing/summary``
            # 的取整一致,跨端点合计才对得上。
            total_cost_usd=float(
                acc.total.quantize(_Dec("0.01"), rounding=_ROUND_HALF_UP),
            ),
            # 两个口径都恒 None。list:日表的 ``cost_by_service`` 是
            # **实付**拆分,与 list 合计同框就把折扣率漏了。invoice:
            # ``cmdb_billing_monthly.cost_by_service`` 的币种在读侧
            # **无法判定** —— 写侧 rebase 成功时只把 ``total_cost_usd``
            # 折成 USD 并把 ``currency`` 标 USD,JSON 拆分留在原生 CNY
            # (handlers.py 的 billing_sync),所以 ``currency=='USD'``
            # 的行里 scale=1、拆分会以 ¥ 数字打着 $ 标发出(实测
            # total 700.0 USD 旁边 ECS 5000.0)。列名之外没有任何 native
            # 总额可用来还原比例,唯一诚实的读法就是不发。服务拆分请走
            # ``/billing/overview`` / ``/billing/summary``(两者的 list
            # 榜从 cost_facts 独立聚合,invoice 榜从日表逐行折算)。
            cost_by_service=None,
            # 数值已是 USD —— 数值与 currency 标签必须同单位(写侧
            # handlers.py 的同一不变量)。
            currency="USD",
            fetched_at=acc.fetched_at,
        ).model_dump(mode="json")
        data.append(payload)
    meta: dict[str, Any] = {
        "measure": resolved_measure,
        # 口径切换时的**覆盖面差异**披露(不是折扣!):invoice 是
        # ``cmdb_billing_monthly`` 的每 integration 一条租户级 provider
        # 汇总;list 是日表逐 scope 行 + ``scope_id IS NULL`` 的未登记
        # 账号对账行之和。对 alicloud-intl 分销主账号这两者按文档就是
        # 两个不同的聚合(alicloud_intl_cloud.py:209-236),日表 4dp /
        # 月表 2dp 也会留尾差。财务切口径看到的差额里含这部分覆盖差,
        # 不可整体当折扣读。
        "aggregate_basis": (
            "tenant_monthly" if resolved_measure == COST_MEASURE_INVOICE
            else "billing_daily_sum"
        ),
    }
    if cost_conv.unconverted_by_currency:
        logger.warning(
            "list_monthly_billing: %s 行无可用汇率,已从 USD 合计中剔除: %s",
            resolved_measure, cost_conv.unconverted_by_currency,
        )
        meta["unconverted_by_currency"] = cost_conv.unconverted_by_currency
    return success_response(data=data, request=request, meta=meta)


@router.get("/billing/summary")
async def get_billing_summary(
    request: Request,
    month: str | None = Query(default=None, description="YYYY-MM; defaults to last full month"),
    integration_id: str | None = Query(
        default=None,
        description=(
            "Restrict the aggregate to a single cloud tenant. When omitted, "
            "summary spans every integration the caller can see."
        ),
    ),
    measure: str | None = Query(
        default=None,
        description=(
            "成本口径:``list``(默认,业务折前成本/研发口径,按月聚合自 "
            "cmdb_cloud_billing_daily)或 ``invoice``(实付,读 "
            "cmdb_billing_monthly,仅 finance / cloud_admin / "
            "global_admin)。"
        ),
    ),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Aggregated multi-cloud summary for the billing page hero card.

    Finance role included in the allowlist alongside cloud-admin.

    口径路由（2026-07 保密红线补口）：本端点是移动端 ``/m/cloud-cost``
    与 ``/m/reports`` 的唯一云花费来源，受众 = ``can_read_cloud`` ∪
    ``can_view_finance``（含 **auditor / management**，两者都在保密圈
    外）。桌面端 ``/billing/monthly`` + ``/billing/overview`` 已按研发
    口径路由，本端点若继续裸读 ``cmdb_billing_monthly``（实付），同一
    个 auditor 桌面看 list、手机看 invoice，**同月同粒度两个口径一相除
    就是公司折扣率** —— 所以默认改为 list，从日表聚合；``invoice`` 仅
    保密圈可请求（无权限 → 403，绝不静默降级）。
    """
    if not (can_read_cloud(user.role_names) or can_view_finance(user.role_names)):
        raise ForbiddenError()
    from app.domains.cmdb.cost_view import (
        COST_MEASURE_INVOICE as _MEASURE_INVOICE,
    )
    from app.domains.cmdb.cost_view import resolve_measure

    try:
        resolved_measure = resolve_measure(measure, user.role_names)
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc
    if resolved_measure is None:
        raise ForbiddenError()
    from datetime import date as _date
    from datetime import timedelta as _td
    from decimal import Decimal as _Decimal
    from types import SimpleNamespace

    from app.domains.cmdb.models import CmdbBillingMonthly

    if month:
        try:
            target = _date.fromisoformat(f"{month}-01")
        except ValueError:
            raise NotFoundError("Billing", month)
    else:
        first_this = _date.today().replace(day=1)
        target = (first_this - _td(days=1)).replace(day=1)

    if integration_id:
        # Narrow to the selected tenant; otherwise aggregate everything.
        try:
            uuid.UUID(integration_id)
        except ValueError:
            raise NotFoundError("Integration", integration_id)

    if resolved_measure == _MEASURE_INVOICE:
        stmt = select(CmdbBillingMonthly).where(
            CmdbBillingMonthly.month == target,
        )
        if integration_id:
            stmt = stmt.where(
                CmdbBillingMonthly.integration_id == integration_id,
            )
        rows: list = list((await db.execute(stmt)).scalars().all())
    else:
        # list 口径:按 (integration, currency) 聚合本月日表。currency
        # 进分组键 —— CN 云行是原生 CNY(列名骗人),混币合计会把 ¥ 当
        # $;下面的逐行折算块负责换算,与 ``/billing/monthly`` 同一套。
        # 服务榜不从这里取(见下方 facts 聚合块)。
        from sqlalchemy import func as _func

        from app.domains.cmdb.cost_view import daily_engineering_amount
        from app.domains.cmdb.models import CmdbCloudBillingDaily

        next_month = (target + _td(days=32)).replace(day=1)
        agg_q = (
            select(
                CmdbCloudBillingDaily.integration_id,
                CmdbCloudBillingDaily.currency,
                _func.coalesce(
                    _func.sum(daily_engineering_amount()), 0,
                ).label("total"),
            )
            .where(
                CmdbCloudBillingDaily.date >= target,
                CmdbCloudBillingDaily.date < next_month,
            )
            .group_by(
                CmdbCloudBillingDaily.integration_id,
                CmdbCloudBillingDaily.currency,
            )
        )
        if integration_id:
            agg_q = agg_q.where(
                CmdbCloudBillingDaily.integration_id == integration_id,
            )
        rows = [
            SimpleNamespace(
                integration_id=r.integration_id,
                total_cost_usd=r.total or 0,
                currency=r.currency,
            )
            for r in (await db.execute(agg_q)).all()
        ]

    # Round-27 pre-launch (2026-05-25): per-row currency-aware totalling.
    # Previously every ``r.total_cost_usd`` was summed regardless of
    # ``r.currency`` — for Aliyun/Tencent rows where the writer stored
    # native CNY in ``total_cost_usd`` with ``currency='CNY'`` (column
    # is misnamed), finance saw a ``$`` headline mixing genuine USD +
    # 7×-inflated CNY-as-USD. Now: one shared ``CloudCostUsdConverter``
    # (same resolution order / cache as ``/billing/monthly`` and
    # ``/billing/overview``); ``to_usd_checked`` gives the per-row
    # "did it convert" signal so an unconvertible integration is
    # DROPPED rather than emitted as a ``$0`` ``by_account`` row (a $0
    # row reads as "this cloud spent nothing this month", which is a
    # different and wrong claim). Disclosed via
    # ``unconverted_by_currency``, same as the sibling endpoints.
    from app.core.fx_rate import CloudCostUsdConverter  # noqa: PLC0415

    cost_conv = CloudCostUsdConverter(db)
    total = _Decimal("0")
    # integration_id → 折算后 USD 行。list 路径同一 integration 可能有
    # 多个币种组(CNY + USD),折算后必须合并成一行,否则移动端账号列表
    # 同一家云出现两条、占比也算错。
    by_account_acc: dict[str, dict] = {}
    service_totals: dict[str, float] = {}
    integrations_cache: dict[str, Integration | None] = {}
    # 任一币种组折不出汇率的 integration —— 整个账号从 by_account 和
    # 合计里剔除(与 ``/billing/monthly`` 的整行不发同一口径:偏低的
    # 数标成完整比缺行更容易误导)。
    partial_integration_ids: set[str] = set()
    for r in rows:
        if r.integration_id not in integrations_cache:
            integ = await db.get(Integration, uuid.UUID(str(r.integration_id)))
            integrations_cache[r.integration_id] = integ
        integ = integrations_cache[r.integration_id]
        if integ is None:
            # 孤儿账单行:integration 被硬删、账单历史留存。与
            # ``/billing/monthly`` / ``/billing/overview`` 一致地跳过,
            # 否则 by_account 里出现一行以 UUID 当名字、provider 为
            # null 的幽灵账号,占比分母也被它污染。
            continue
        cost_usd, converted = await cost_conv.to_usd_checked(
            r.total_cost_usd, r.currency, str(r.integration_id),
        )
        if not converted:
            partial_integration_ids.add(str(r.integration_id))
            continue
        cost = _Decimal(str(cost_usd))
        total += cost
        acct = by_account_acc.setdefault(str(r.integration_id), {
            "integration_id": str(r.integration_id),
            "integration_name": integ.name,
            "provider": integ.provider,
            "total_cost_usd": 0.0,
            # 数值已折算成 USD —— 标签必须跟着单位走。
            "currency": "USD",
        })
        acct["total_cost_usd"] += float(cost)

    for bad_id in partial_integration_ids:
        dropped = by_account_acc.pop(bad_id, None)
        if dropped is not None:
            total -= _Decimal(str(dropped["total_cost_usd"]))

    # 服务榜:**两个口径都**从 ``cmdb_cost_facts`` 按 ``fact_amount_expr``
    # 聚合(list = COALESCE(amount_usd_list, amount_usd),invoice =
    # amount_usd)。理由:
    #   * list —— 日表 / 月表的 ``cost_by_service`` 是实付拆分,与 list
    #     合计同框就把 (服务 × 月) 粒度的折扣率漏了。
    #   * invoice —— ``cmdb_billing_monthly.cost_by_service`` 的币种在
    #     读侧不可判定:写侧 rebase 只折 ``total_cost_usd`` 并把
    #     ``currency`` 标成 USD,JSON 拆分留在原生 CNY,于是按行
    #     scale=1 缩放会把 ¥ 数字打着 $ 标发出(实测 total 700.0 USD
    #     旁边 ECS 5000.0)。facts 的 ``amount_usd*`` 是入库时就折算好
    #     的真 USD,没有这个歧义。
    # 限制在 hero 合计实际覆盖的 integration 上 —— 此前 facts 查询无论
    # 是否传 ``integration_id`` 都扫全部租户,榜单跨了合计没覆盖的云。
    #
    # 覆盖面提示:未接账单明细的云(腾讯 / 金山 / UCloud、Azure 待
    # Cost Exports)没有 facts 行,服务榜会比合计窄;移动端的服务
    # 卡片对空列表是**不渲染**(不是空壳),hero + 账号列表照常在,
    # 不构成死胡同页。
    counted_integration_ids = set(by_account_acc)
    if counted_integration_ids:
        from sqlalchemy import func as _svc_func

        from app.domains.cmdb.cost_view import fact_amount_expr
        from app.domains.cmdb.models import CmdbCostFact

        _next_month = (target + _td(days=32)).replace(day=1)
        svc_q = (
            select(
                CmdbCostFact.service,
                _svc_func.coalesce(
                    _svc_func.sum(fact_amount_expr(resolved_measure)), 0,
                ).label("usd"),
            )
            .where(
                CmdbCostFact.date >= target,
                CmdbCostFact.date < _next_month,
                CmdbCostFact.integration_id.in_(counted_integration_ids),
            )
            .group_by(CmdbCostFact.service)
        )
        for r in (await db.execute(svc_q)).all():
            if not r.service:
                continue
            service_totals[r.service] = (
                service_totals.get(r.service, 0) + float(r.usd or 0)
            )

    # Top-N services with the same SP-pin rule as ``/billing/overview``
    # — see that endpoint for the rationale. Linked-account scopes net
    # to a negative ``Savings Plans`` entry that would otherwise drop
    # off the top-10 cliff.
    _SP_SERVICE_LABEL = "Savings Plans"
    positive_services = [
        {"service": k, "cost_usd": v}
        for k, v in service_totals.items()
        if v > 0 and k != _SP_SERVICE_LABEL
    ]
    top_services = sorted(
        positive_services, key=lambda x: x["cost_usd"], reverse=True,
    )[:10]
    sp_total = service_totals.get(_SP_SERVICE_LABEL)
    if sp_total is not None and sp_total != 0:
        sp_row = {
            "service": _SP_SERVICE_LABEL,
            "cost_usd": sp_total,
            "is_discount": sp_total < 0,
        }
        insert_at = 0
        sp_abs = abs(sp_total)
        for i, row in enumerate(top_services):
            if abs(row["cost_usd"]) >= sp_abs:
                insert_at = i + 1
            else:
                break
        top_services = top_services[:insert_at] + [sp_row] + top_services[insert_at:]
        top_services = top_services[:11]
    summary = BillingSummary(
        month=target,
        total_cost_usd=float(total),
        by_account=sorted(
            by_account_acc.values(),
            key=lambda x: x["total_cost_usd"], reverse=True,
        ),
        top_services=top_services,
    )
    payload = summary.model_dump()
    # 口径标识:移动端据此渲染徽标,并保证它与桌面账单页同框同口径。
    payload["measure"] = resolved_measure
    # 覆盖面基准 —— 与 ``/billing/monthly`` 同语义:invoice 是租户级
    # provider 月汇总,list 是日表逐 scope 行 + 未登记账号对账行之和。
    # 两者对 alicloud-intl 分销主账号按文档就是不同聚合,切口径看到的
    # 差额含覆盖差,不可整体当折扣读。
    payload["aggregate_basis"] = (
        "tenant_monthly" if resolved_measure == _MEASURE_INVOICE
        else "billing_daily_sum"
    )
    # 服务榜与 hero 合计不同源(facts vs 日表/月表)—— 榜单之和 < 合计
    # 时读者据此判断那是覆盖差而不是折扣。
    payload["top_services_basis"] = "cost_facts_list" if (
        resolved_measure != _MEASURE_INVOICE
    ) else "cost_facts_invoice"
    # Surface FX-missing native amounts so the FE can render a "X 行未
    # 折算 USD" banner. Empty = clean totals.
    if cost_conv.unconverted_by_currency:
        logger.warning(
            "get_billing_summary: %s 行无可用汇率,相关账号已从合计中"
            "剔除: %s",
            resolved_measure, cost_conv.unconverted_by_currency,
        )
        payload["unconverted_by_currency"] = cost_conv.unconverted_by_currency
    return success_response(data=payload, request=request)


# ---------------------------------------------------------------------------
# Billing — daily granularity (added 2026-04-29 for the redesigned page)
# ---------------------------------------------------------------------------
#
# The original page only consumed monthly aggregates from
# ``cmdb_billing_monthly``. The redesigned billing page asks "how much did
# each cloud cost on each day in this period?" which requires the daily
# table that the ``cmdb.billing_daily_sync`` cron has been populating
# since migration 055. These endpoints expose it.


def _parse_iso_date(label: str, raw: str) -> date:
    """Tight ISO date parser that surfaces a 422 instead of a 500 when
    the client sends garbage. Centralized so all billing endpoints
    speak the same dialect."""
    try:
        return date.fromisoformat(raw)
    except ValueError:
        from app.core.exceptions import AppException
        raise AppException(
            code="VALIDATION_ERROR",
            message=f"`{label}` must be ISO-8601 (YYYY-MM-DD), got {raw!r}",
            status_code=422,
            field=label,
        )


# Hard cap on the requested window. The overview endpoint walks every
# row in [prev_start, end] which is 2× the requested span, and joins
# against integrations + cost_by_service JSON. At our current shape
# (~5 cloud tenants × ~20 scopes × 365 days × 2 = 73 000 rows) the
# request still completes in well under a second; capping at 18
# months keeps even an adversarial range (YoY analysis is the
# longest legit use case) bounded so a stray ``from=1970-01-01`` can't
# OOM the worker. The frontend "近 12 月" preset is 365 days, well
# within the cap.
_BILLING_RANGE_MAX_DAYS = 18 * 31  # 558 days, ~18 months


def _resolve_daily_range(
    from_q: str | None, to_q: str | None,
) -> tuple[date, date]:
    """Default the range to the trailing 30 days (inclusive) when the
    client omits one or both ends. ``to`` defaults to *yesterday* — the
    daily-sync cron runs at 06:00 UTC so today's row isn't reliable
    until then; surfacing a partial today would mislead the trend.

    Hard-caps the resolved span at ``_BILLING_RANGE_MAX_DAYS`` so a
    request like ``from=1970-01-01&to=2099-12-31`` 422s instead of
    OOMing the worker on the JOIN-and-walk in ``/billing/overview``.
    """
    from datetime import timedelta as _td

    today = datetime.now(UTC).date()
    if to_q:
        end = _parse_iso_date("to", to_q)
    else:
        end = today - _td(days=1)
    if from_q:
        start = _parse_iso_date("from", from_q)
    else:
        start = end - _td(days=29)  # inclusive 30-day window
    if start > end:
        from app.core.exceptions import AppException
        raise AppException(
            code="VALIDATION_ERROR",
            message="`from` must be on or before `to`.",
            status_code=422,
        )
    span_days = (end - start).days + 1
    if span_days > _BILLING_RANGE_MAX_DAYS:
        from app.core.exceptions import AppException
        raise AppException(
            code="VALIDATION_ERROR",
            message=(
                f"Range too large: {span_days} days requested, max is "
                f"{_BILLING_RANGE_MAX_DAYS} (~18 months). Narrow the window."
            ),
            status_code=422,
        )
    return start, end


@router.get("/billing/overview")
async def get_billing_overview(
    request: Request,
    from_: str | None = Query(default=None, alias="from"),
    to: str | None = Query(default=None),
    integration_id: str | None = Query(default=None),
    measure: str | None = Query(
        default=None,
        description=(
            "成本口径:``list``(默认,业务折前成本/研发口径)或 "
            "``invoice``(实付,仅 finance / cloud_admin / global_admin)。"
        ),
    ),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Single-shot range overview powering the redesigned billing page.

    Consolidates totals, per-provider breakdown (with prior-period delta),
    by-account table, top services, and the dense daily series into one
    response so the page renders in a single round-trip. The prior-period
    comparison uses an immediately-preceding window of identical length
    — that's the only definition that survives quarter-end / month-end
    boundaries cleanly (a "same calendar period last month" comparison
    would lie about a 31-day → 30-day month transition).
    """
    if not (can_read_cloud(user.role_names) or can_view_finance(user.role_names)):
        raise ForbiddenError()
    # 口径路由:默认研发口径;invoice 仅保密圈,无权限 → 403。
    from app.domains.cmdb.cost_view import resolve_measure

    try:
        resolved_measure = resolve_measure(measure, user.role_names)
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc
    if resolved_measure is None:
        raise ForbiddenError()
    from datetime import timedelta as _td

    from app.domains.cmdb.models import CmdbCloudBillingDaily

    start, end = _resolve_daily_range(from_, to)
    days = (end - start).days + 1
    prev_end = start - _td(days=1)
    prev_start = prev_end - _td(days=days - 1)

    if integration_id:
        try:
            uuid.UUID(integration_id)
        except ValueError:
            raise NotFoundError("Integration", integration_id)

    # Single query covers BOTH the current window and the prior-period
    # comparison so the prior-period total comes from the same scan
    # rather than a second round-trip. We tag each row by which window
    # it belongs to via a CASE expression.
    from app.domains.cmdb.cost_view import daily_amount_expr

    base = select(
        CmdbCloudBillingDaily.date,
        CmdbCloudBillingDaily.integration_id,
        CmdbCloudBillingDaily.scope_id,
        CmdbCloudBillingDaily.linked_account_id,
        # 2026-07 成本口径改造:总览账单 tab(can_read_cloud 受众含
        # 研发管理/审计)默认研发口径;invoice 仅保密圈(上方 403 闸)。
        daily_amount_expr(resolved_measure).label("total_cost_usd"),
        # 这两个 JSON 列**恒为实付口径**(驱动按账单原样落盘,日表没有
        # 对应的 list 列)。只有 invoice 路径消费它们;list 路径下
        # ``top_services`` 改从 ``cmdb_cost_facts`` 的 list 列独立聚合、
        # ``owner_breakdown`` 直接不发 —— 否则同一响应里 list 合计 +
        # 实付拆分同框,(服务|owner-tag × 窗口) 粒度的折扣率就能相除
        # 导出(实测:measure=list total 1000.0 旁边 top_services
        # [{EC2: 300.0}, {S3: 100.0}] → 比值 0.4)。
        CmdbCloudBillingDaily.cost_by_service,
        CmdbCloudBillingDaily.owner_breakdown,
        # Round-27 pre-launch (2026-05-25): pull currency so the
        # CNY-as-USD column-name confusion can be corrected at read
        # time. Aliyun/Tencent daily writers store native CNY in
        # ``total_cost_usd`` with ``currency='CNY'`` — without this
        # column the overview hero/provider/account/daily totals
        # silently 7×-inflated CNY values.
        CmdbCloudBillingDaily.currency,
    ).where(
        CmdbCloudBillingDaily.date >= prev_start,
        CmdbCloudBillingDaily.date <= end,
    )
    if integration_id:
        base = base.where(CmdbCloudBillingDaily.integration_id == integration_id)

    raw_rows = (await db.execute(base)).all()

    # Resolve integration metadata once.
    integration_ids = {str(r.integration_id) for r in raw_rows}
    integrations: dict[str, Integration] = {}
    if integration_ids:
        for integ in (
            await db.execute(
                select(Integration).where(
                    Integration.id.in_([uuid.UUID(i) for i in integration_ids]),
                )
            )
        ).scalars().all():
            integrations[str(integ.id)] = integ

    # Walk the rows once, accumulating into the response shape.
    cur_total = 0.0
    prev_total = 0.0
    cur_unregistered = 0.0  # subset of cur_total from scope_id=NULL rows
    cur_by_provider: dict[str, float] = {}
    prev_by_provider: dict[str, float] = {}
    cur_provider_accounts: dict[str, set[str]] = {}
    cur_by_account: dict[str, float] = {}
    # Unregistered accounts keyed by (integration_id, linked_account_id).
    cur_unregistered_accounts: dict[tuple[str, str], float] = {}
    cur_service_totals: dict[str, float] = {}
    cur_owner_totals: dict[str, float] = {}
    daily_by_provider: dict[date, dict[str, float]] = {}
    # 日表的 ``cost_by_service`` / ``owner_breakdown`` 只在 invoice 口径
    # 下消费(见 SELECT 处注释)。list 口径的服务榜在循环后从
    # ``cmdb_cost_facts`` 重算,owner 拆分不发。
    from app.domains.cmdb.cost_view import COST_MEASURE_INVOICE as _MEAS_INV

    wants_invoice_companions = resolved_measure == _MEAS_INV
    # 参与合计的 integration 集合 —— list 口径的 facts 服务榜必须限制在
    # 它上面,否则榜单跨了 hero 合计没覆盖的租户(占比分母/榜单口径都
    # 会与合计脱节)。
    contributing_integration_ids: set[str] = set()

    # Round-27 pre-launch: per-integration CNY→USD rate cache. Resolved
    # lazily on first encounter of a CNY row for that integration.
    from app.core import fx_rate as _fx_rate  # noqa: PLC0415
    cny_rate_cache: dict[str, float] = {}
    overview_unconverted_by_currency: dict[str, float] = {}

    async def _row_to_usd(
        raw_amount: float,
        row_currency: str,
        integration_id_str: str,
        integration_obj: Integration | None,
    ) -> tuple[float, bool]:
        """Convert raw billing-daily amount to USD according to row
        currency. Returns ``(usd, converted)``.

        ``converted is False`` → no FX rate covers the row; the caller
        must skip the row entirely rather than fold a ``0.0`` into the
        aggregates (a $0 ``by_account`` row reads as "this tenant spent
        nothing this window"). The native amount lands in
        ``overview_unconverted_by_currency`` so the FE renders a banner.
        """
        if row_currency == "USD":
            return raw_amount, True
        if row_currency == "CNY":
            if integration_id_str not in cny_rate_cache:
                try:
                    cny_rate_cache[integration_id_str] = (
                        await _fx_rate.resolve_cny_per_usd(
                            db,
                            integration_config=(
                                integration_obj.config_public
                                if integration_obj else None
                            ),
                        )
                    )
                except RuntimeError:
                    cny_rate_cache[integration_id_str] = 0.0
            rate = cny_rate_cache[integration_id_str]
            if rate > 0:
                return raw_amount / rate, True
            overview_unconverted_by_currency["CNY"] = (
                overview_unconverted_by_currency.get("CNY", 0.0)
                + raw_amount
            )
            return 0.0, False
        overview_unconverted_by_currency[row_currency] = (
            overview_unconverted_by_currency.get(row_currency, 0.0)
            + raw_amount
        )
        return 0.0, False

    for r in raw_rows:
        raw_amount = float(r.total_cost_usd or 0)
        integ = integrations.get(str(r.integration_id))
        if integ is None:
            # Orphan billing row — the integration row was hard-deleted
            # but billing history was kept. Skip so the page's totals,
            # provider cards, and prior-period comparison all reflect
            # the same attributable row set. The raw table is still
            # there for forensic queries; the API contract is "what
            # we can show honestly". ``integ.provider`` is column-level
            # NOT NULL on the integrations table, so there's no second
            # branch to handle.
            continue
        row_currency = (r.currency or "USD").upper()
        amount, converted = await _row_to_usd(
            raw_amount, row_currency, str(r.integration_id), integ,
        )
        if not converted:
            # No FX rate → the row has no USD value. Skip it outright;
            # folding a 0.0 in would put a fabricated $0 row into
            # ``by_account`` / ``by_provider`` for a tenant that did
            # spend money. Disclosed via ``unconverted_by_currency``.
            continue
        # Per-service / per-owner breakdowns are in native currency too —
        # apply the same scale factor.
        scale = (amount / raw_amount) if raw_amount else 1.0
        provider = integ.provider
        in_current = start <= r.date <= end
        is_unregistered = r.scope_id is None

        if in_current:
            cur_total += amount
            contributing_integration_ids.add(str(r.integration_id))
            cur_by_provider[provider] = cur_by_provider.get(provider, 0) + amount
            day_bucket = daily_by_provider.setdefault(r.date, {})
            day_bucket[provider] = day_bucket.get(provider, 0) + amount

            if is_unregistered:
                # Unregistered linked account — track separately so the
                # caller can surface the registration gap to the admin.
                cur_unregistered += amount
                acct_id = r.linked_account_id or ""
                key = (str(r.integration_id), acct_id)
                cur_unregistered_accounts[key] = (
                    cur_unregistered_accounts.get(key, 0) + amount
                )
            else:
                # Registered scope — contributes to per-integration by_account.
                cur_provider_accounts.setdefault(provider, set()).add(
                    str(r.integration_id)
                )
                cur_by_account[str(r.integration_id)] = (
                    cur_by_account.get(str(r.integration_id), 0) + amount
                )
                if wants_invoice_companions:
                    for svc, v in (r.cost_by_service or {}).items():
                        # Round-27 pre-launch: convert per-service slice
                        # using the same row-scale so CNY/USD rows align
                        # at the bucket level too.
                        cur_service_totals[svc] = (
                            cur_service_totals.get(svc, 0) + float(v) * scale
                        )

            # Owner breakdown — merge across all scopes / providers.
            # invoice 口径专属(见 SELECT 处注释:JSON 列无 list 对应)。
            if wants_invoice_companions:
                for owner, v in (r.owner_breakdown or {}).items():
                    cur_owner_totals[owner] = (
                        cur_owner_totals.get(owner, 0) + float(v) * scale
                    )
        else:
            prev_total += amount
            prev_by_provider[provider] = prev_by_provider.get(provider, 0) + amount

    if not wants_invoice_companions:
        # list 口径的服务榜:从 ``cmdb_cost_facts`` 的 list 列独立聚合
        # (``fact_engineering_amount`` = COALESCE(list, invoice)),与
        # ``/billing/summary`` 同一做法。facts 的 ``amount_usd*`` 本就
        # 是 USD,不再折算。限制在 ``contributing_integration_ids`` ——
        # 榜单与 hero 合计必须覆盖同一批租户。
        #
        # 覆盖面提示:未接账单明细的云(腾讯 / 金山 / UCloud、Azure 待
        # Cost Exports)没有 facts 行,服务榜比合计窄 —— 这是覆盖差,
        # 不是折扣。页面其余面板(hero / provider / by_account / daily)
        # 照常渲染,不构成死胡同。
        from sqlalchemy import func as _svc_func

        from app.domains.cmdb.cost_view import fact_engineering_amount
        from app.domains.cmdb.models import CmdbCostFact

        if contributing_integration_ids:
            svc_q = (
                select(
                    CmdbCostFact.service,
                    _svc_func.coalesce(
                        _svc_func.sum(fact_engineering_amount()), 0,
                    ).label("usd"),
                )
                .where(
                    CmdbCostFact.date >= start,
                    CmdbCostFact.date <= end,
                    CmdbCostFact.integration_id.in_(
                        contributing_integration_ids,
                    ),
                )
                .group_by(CmdbCostFact.service)
            )
            for fr in (await db.execute(svc_q)).all():
                if not fr.service:
                    continue
                cur_service_totals[fr.service] = (
                    cur_service_totals.get(fr.service, 0) + float(fr.usd or 0)
                )

    # Backfill missing days so the chart's x-axis is contiguous.
    cursor = start
    while cursor <= end:
        daily_by_provider.setdefault(cursor, {})
        cursor += _td(days=1)

    daily_points = [
        BillingDailyPoint(
            date=d,
            by_provider=daily_by_provider[d],
            total=sum(daily_by_provider[d].values()),
        )
        for d in sorted(daily_by_provider.keys())
    ]

    def _pct(cur: float, prev: float) -> float | None:
        if prev <= 0:
            return None
        return ((cur - prev) / prev) * 100

    by_provider_rows = [
        BillingProviderBreakdown(
            provider=p,
            total_cost_usd=cur_by_provider[p],
            prev_total_cost_usd=prev_by_provider.get(p, 0.0),
            delta_pct=_pct(cur_by_provider[p], prev_by_provider.get(p, 0.0)),
            share_pct=(cur_by_provider[p] / cur_total * 100) if cur_total > 0 else 0.0,
            account_count=len(cur_provider_accounts.get(p, set())),
        )
        for p in sorted(cur_by_provider, key=lambda k: cur_by_provider[k], reverse=True)
    ]

    # Every ``iid`` in ``cur_by_account`` is guaranteed in
    # ``integrations`` because the accumulator loop above skips orphan
    # rows up-front. No defensive fallback needed for the lookups.
    by_account_rows_base = [
        {
            "integration_id": iid,
            "integration_name": integrations[iid].name,
            "provider": integrations[iid].provider,
            "total_cost_usd": amt,
            "share_pct": (amt / cur_total * 100) if cur_total > 0 else 0.0,
            "is_unregistered": False,
        }
        for iid, amt in cur_by_account.items()
    ]
    # Append unregistered-account rows so admins see the attribution gap.
    # ``linked_account_id`` surfaces the raw cloud account ID so the admin
    # knows which account to register.
    for (iid, acct_id), amt in cur_unregistered_accounts.items():
        integ = integrations.get(iid)
        if integ is None:
            continue
        by_account_rows_base.append({
            "integration_id": iid,
            "integration_name": integ.name,
            "provider": integ.provider,
            "total_cost_usd": amt,
            "share_pct": (amt / cur_total * 100) if cur_total > 0 else 0.0,
            "is_unregistered": True,
            "linked_account_id": acct_id,
        })
    by_account_rows = sorted(
        by_account_rows_base,
        key=lambda x: x["total_cost_usd"],
        reverse=True,
    )

    # Top-N services: positive entries sorted by cost desc, sliced to 10.
    # The ``Savings Plans`` bucket (introduced PR #339 — captures the AWS
    # CUR SP discount on linked accounts as a negative line, or the SP
    # commitment fee on payer accounts as a positive line) is PINNED as
    # an additional row regardless of where it lands by absolute cost.
    # Without pinning, on linked accounts the SP row is the MOST negative
    # entry and sorts to the bottom — it would drop off the top-10 and
    # leave finance unable to explain the ~30% gap between gross service
    # cost and the reported total. The pin guarantees the discount line
    # is always visible alongside the gross-service ranking.
    _SP_SERVICE_LABEL = "Savings Plans"
    positive_services = [
        {"service": k, "cost_usd": v}
        for k, v in cur_service_totals.items()
        if v > 0 and k != _SP_SERVICE_LABEL
    ]
    top_services = sorted(
        positive_services, key=lambda x: x["cost_usd"], reverse=True,
    )[:10]
    sp_total = cur_service_totals.get(_SP_SERVICE_LABEL)
    if sp_total is not None and sp_total != 0:
        # Order-aware insert: SP row goes at the position its absolute
        # value would land if sorted by |cost| — typically near the top
        # for any SP-active tenant.
        sp_row = {
            "service": _SP_SERVICE_LABEL,
            "cost_usd": sp_total,
            "is_discount": sp_total < 0,
        }
        insert_at = 0
        sp_abs = abs(sp_total)
        for i, row in enumerate(top_services):
            if abs(row["cost_usd"]) >= sp_abs:
                insert_at = i + 1
            else:
                break
        top_services = top_services[:insert_at] + [sp_row] + top_services[insert_at:]
        top_services = top_services[:11]  # 10 services + the pinned SP row

    overview = BillingOverview(
        from_date=start,
        to_date=end,
        days=days,
        total_cost_usd=cur_total,
        prev_total_cost_usd=prev_total,
        delta_pct=_pct(cur_total, prev_total),
        avg_daily_cost_usd=(cur_total / days) if days > 0 else 0.0,
        unregistered_cost_usd=cur_unregistered,
        by_provider=by_provider_rows,
        by_account=by_account_rows,
        top_services=top_services,
        owner_breakdown=cur_owner_totals,
        daily=daily_points,
    )
    # ``by_alias=True`` so ``from``/``to`` come out under their real
    # names (the model field is ``from_date``/``to_date`` because
    # ``from`` is a Python keyword).
    payload = overview.model_dump(mode="json", by_alias=True)
    # 口径标识:FE 据此渲染徽标/联动页级切换。
    payload["measure"] = resolved_measure
    # 服务榜的来源表(与 hero 合计不同源):list 走 cost_facts 的 list
    # 列,invoice 走日表的 ``cost_by_service``。榜单之和 < 合计时读者
    # 据此判断那是覆盖差而不是折扣。``owner_breakdown`` 在 list 口径下
    # 恒空 —— 它只有实付一个版本(cloud owner-tag 键空间在 cost_facts
    # 里没有对应列),与 list 合计同框即可按 owner-tag 粒度相除。
    payload["top_services_basis"] = (
        "billing_daily_cost_by_service" if wants_invoice_companions
        else "cost_facts_list"
    )
    # Round-27 pre-launch: surface FX-missing native sums so the
    # FE can render a "X 行未折算 USD" banner. Empty = clean totals.
    if overview_unconverted_by_currency:
        payload["unconverted_by_currency"] = overview_unconverted_by_currency
    return success_response(
        data=payload,
        request=request,
    )


# ---------------------------------------------------------------------------
# Billing — CSV export (added 2026-04-29)
# ---------------------------------------------------------------------------
#
# Two flat shapes both keyed off the same overview math:
#   * ``kind=daily``      — one row per (date, integration), suitable for
#                           Excel pivots / Looker Studio drag-drop.
#   * ``kind=by_account`` — one row per integration with period totals,
#                           share %, prior-period spend (when available).
#
# Both gate on ``can_read_cloud OR can_view_finance`` like the JSON
# endpoints. We stream via ``StreamingResponse`` so a 90-day × 20-account
# export doesn't materialise the whole CSV in memory.

_CSV_KINDS = frozenset({"daily", "by_account"})


@router.get("/billing/export.csv")
async def export_billing_csv(
    request: Request,
    kind: str = Query(default="daily", description="``daily`` or ``by_account``."),
    from_: str | None = Query(default=None, alias="from"),
    to: str | None = Query(default=None),
    integration_id: str | None = Query(default=None),
    measure: str | None = Query(
        default=None,
        description=(
            "成本口径:``list``(默认,业务折前成本/研发口径)或 "
            "``invoice``(实付,仅 finance / cloud_admin / global_admin)。"
        ),
    ),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Stream a CSV export of the requested billing slice.

    The two shapes intentionally don't share a header — analyst tools
    don't merge mismatched schemas cleanly, and "one CSV with both"
    would force a sentinel-row design that's confusing to read.
    Pick the shape you need; the page UI exposes both as separate
    download buttons.

    口径路由：与它导出的那个页面（``/billing/overview`` +
    ``/billing/monthly``）同一套。默认 ``list``；``invoice`` 仅保密圈，
    无权限 → 403。导出是最容易被拿去做表的面 —— 若它继续吐实付而页面
    吐 list，一个 auditor 下载一次 CSV 就把折扣率固化进电子表格了。
    """
    if not (can_read_cloud(user.role_names) or can_view_finance(user.role_names)):
        raise ForbiddenError()
    from app.domains.cmdb.cost_view import resolve_measure

    try:
        resolved_measure = resolve_measure(measure, user.role_names)
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc
    if resolved_measure is None:
        raise ForbiddenError()
    if kind not in _CSV_KINDS:
        from app.core.exceptions import AppException
        raise AppException(
            code="VALIDATION_ERROR",
            message=f"`kind` must be one of {sorted(_CSV_KINDS)}, got {kind!r}",
            status_code=422,
            field="kind",
        )

    from app.domains.cmdb.models import CmdbCloudBillingDaily

    start, end = _resolve_daily_range(from_, to)

    if integration_id:
        try:
            uuid.UUID(integration_id)
        except ValueError:
            raise NotFoundError("Integration", integration_id)

    from app.domains.cmdb.cost_view import daily_amount_expr

    base = (
        select(
            CmdbCloudBillingDaily.date,
            CmdbCloudBillingDaily.integration_id,
            daily_amount_expr(resolved_measure).label("total_cost_usd"),
            CmdbCloudBillingDaily.currency,
        )
        .where(CmdbCloudBillingDaily.date >= start)
        .where(CmdbCloudBillingDaily.date <= end)
    )
    if integration_id:
        base = base.where(CmdbCloudBillingDaily.integration_id == integration_id)

    rows = list((await db.execute(base)).all())

    integration_ids = {str(r.integration_id) for r in rows}
    integrations: dict[str, Integration] = {}
    if integration_ids:
        for integ in (
            await db.execute(
                select(Integration).where(
                    Integration.id.in_([uuid.UUID(i) for i in integration_ids]),
                )
            )
        ).scalars().all():
            integrations[str(integ.id)] = integ

    buf = io.StringIO()
    writer = csv.writer(buf)

    # Filter out orphan rows (integration row hard-deleted) BEFORE
    # aggregation so the share-percent denominators reflect the same
    # row set the analyst sees, not phantoms. Same reasoning as the
    # JSON endpoints — emitting "<uuid>" as the integration name in
    # an export would just confuse downstream pivots.
    attributable_rows = [r for r in rows if str(r.integration_id) in integrations]

    # 逐行 CNY→USD 折算,然后才聚合。此前两个 shape 都裸加 ``total_cost_usd``
    # 并把 ``currency`` 列硬编码成 "USD" —— CN 云的行是原生 CNY,导出的
    # 数字 ~7× 虚高却打着 USD 的标,财务拿去做表就直接错了(而且与同页
    # 的 ``/billing/overview`` 数字不一致)。同一个 ``CloudCostUsdConverter``。
    #
    # 折不出汇率的行**整行不进 CSV**:此前它们贡献 0,于是导出里出现
    # ``2026-07-28,…,aliyun,0.0000,USD`` —— 一个 ¥5000 的真实花费在
    # 电子表格里读成「这天这家云没花钱」,而唯一披露只有服务端日志
    # (CSV 没有 meta 可承载)。改为:金额行只含折算成功的行,
    # 未折算的原生额以尾部注释行披露在同一个文件里 —— 首列以 ``#``
    # 开头(CSV 的注释约定),金额列写死非数值 ``UNCONVERTED``,
    # 电子表格对金额列做 SUM 时既不会加 0 也不会加原生数。
    from app.core.fx_rate import CloudCostUsdConverter  # noqa: PLC0415

    csv_conv = CloudCostUsdConverter(db)
    usd_rows: list[tuple[date, str, float]] = []
    for r in attributable_rows:
        amount_usd, converted = await csv_conv.to_usd_checked(
            r.total_cost_usd, r.currency, str(r.integration_id),
        )
        if not converted:
            continue
        usd_rows.append((r.date, str(r.integration_id), amount_usd))
    if csv_conv.unconverted_by_currency:
        logger.warning(
            "export_billing_csv: %s 行无可用汇率,已从导出金额中剔除: %s",
            resolved_measure, csv_conv.unconverted_by_currency,
        )

    def _write_unconverted_footer(amount_col_index: int, width: int) -> None:
        """尾部披露行:未折算的原生额,一行一个币种。

        首列 ``# UNCONVERTED`` 标明这不是数据行;金额列写死字符串
        ``UNCONVERTED`` 而不是数字,所以对金额列求和的表格公式拿不到
        任何值(0 也拿不到)。原生额与币种放在末列的说明文字里。
        """
        for cur, native in sorted(csv_conv.unconverted_by_currency.items()):
            row: list[Any] = [""] * width
            row[0] = "# UNCONVERTED"
            row[amount_col_index] = "UNCONVERTED"
            row[width - 1] = (
                f"{native:.4f} {cur} excluded — no CNY/USD rate available "
                f"(platform setting ai_spending.cny_to_usd_rate missing, "
                f"stale, or out of band); re-export after ops restores it."
            )
            writer.writerow(_csv_safe_row(row))

    if kind == "daily":
        writer.writerow([
            "date", "integration_id", "integration_name", "provider",
            "total_cost_usd", "currency",
        ])
        # Sum (integration, day) inside the loop so a tenant with 20
        # scopes still produces one row per day rather than 20.
        agg: dict[tuple[date, str], float] = {}
        for day, iid_, amount_usd in usd_rows:
            key = (day, iid_)
            agg[key] = agg.get(key, 0.0) + amount_usd
        for (d, iid), amount in sorted(agg.items()):
            integ = integrations[iid]
            writer.writerow(_csv_safe_row([
                d.isoformat(),
                iid,
                integ.name,
                integ.provider or "",
                f"{amount:.4f}",
                "USD",
            ]))
        # ``total_cost_usd`` 是第 5 列(index 4),共 6 列。
        _write_unconverted_footer(4, 6)
    else:  # kind == "by_account"
        writer.writerow([
            "integration_id", "integration_name", "provider",
            "from", "to", "days", "total_cost_usd", "share_pct",
        ])
        per_account: dict[str, float] = {}
        for _day, iid, amount_usd in usd_rows:
            per_account[iid] = per_account.get(iid, 0.0) + amount_usd
        total = sum(per_account.values())
        days = (end - start).days + 1
        for iid, amount in sorted(per_account.items(), key=lambda kv: kv[1], reverse=True):
            integ = integrations[iid]
            share = (amount / total * 100) if total > 0 else 0.0
            writer.writerow(_csv_safe_row([
                iid,
                integ.name,
                integ.provider or "",
                start.isoformat(),
                end.isoformat(),
                days,
                f"{amount:.4f}",
                f"{share:.2f}",
            ]))
        # ``total_cost_usd`` 是第 7 列(index 6),共 8 列。
        _write_unconverted_footer(6, 8)

    buf.seek(0)
    payload = buf.getvalue()
    filename = f"helix-billing-{kind}-{start.isoformat()}-to-{end.isoformat()}.csv"
    return StreamingResponse(
        iter([payload]),
        media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


# ── Server owner half-yearly confirmation (migration 213) ─────────
#
# Three endpoints, all owner-scoped (the actor must be the owner of
# the confirmation row — no admin override; admins handle escalations
# through the audit log + targeted server edits, not by clicking
# someone else's confirmation card).

class ServerConfirmationResponse(BaseModel):
    # ``Literal`` rejects unknown kinds at the schema layer (422) so
    # the route's ``else`` branch is only reachable for genuinely
    # impossible inputs. Documents the contract in OpenAPI too.
    kind: Literal["confirmed_all", "partial"] = Field(
        ..., description="Response kind."
    )
    # Cap the list length so a million-id payload doesn't take the
    # request path through O(N) parsing + intersection before the
    # snapshot-set filter would have shrunk it. 10k is well above the
    # any-real-owner-can-hold-this-many-servers ceiling.
    server_ids: list[str] | None = Field(
        None,
        max_length=10000,
        description=(
            "Required when kind='partial' — the subset of snapshot "
            "server_ids the owner is confirming this click."
        ),
    )


@router.get("/server-confirmations/mine")
async def list_my_server_confirmations(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Pending server-inventory confirmations for the signed-in owner.

    Surfaces the un-responded cards as in-app TODO. Returns the
    snapshot server_ids + a hydrated subset (name / region / status /
    provider) so the management drawer can render without a second
    fetch per server.
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import CmdbServer
    from app.domains.cmdb.server_owner_confirmation_service import (
        ServerOwnerConfirmationService,
    )

    service = ServerOwnerConfirmationService(db)
    rows = await service.list_pending_for_owner(str(user.id))
    if not rows:
        return success_response(data=[], request=request)

    all_server_ids: set[str] = set()
    for row in rows:
        all_server_ids.update(row.server_ids)

    server_meta: dict[str, dict[str, Any]] = {}
    if all_server_ids:
        # Materialize once — scalars() exhausts the result row, so we
        # collect into a list and iterate twice (integration meta
        # resolution + hydration loop) without re-issuing the SELECT.
        # Soft-deleted servers are excluded — a snapshot entry that a
        # later provider sync soft-deleted must not render as a
        # confirmable row in the drawer (the daily reconcile cron
        # drops it from the snapshot; this filter covers the window
        # between the sync and the next reconcile tick).
        result = await db.execute(
            select(CmdbServer).where(
                CmdbServer.id.in_(list(all_server_ids)),
                CmdbServer.deleted_at.is_(None),
            ),
        )
        servers = list(result.scalars().all())
        integ_meta = await _resolve_integration_meta(
            db, [str(s.integration_id) for s in servers],
        )
        for s in servers:
            meta = integ_meta.get(str(s.integration_id), {}) if integ_meta else {}
            server_meta[str(s.id)] = {
                "id": str(s.id),
                "instance_id": s.instance_id,
                "name": s.name,
                "region": s.region,
                "status": s.status,
                "provider": meta.get("provider"),
                "last_owner_confirmed_at": (
                    s.last_owner_confirmed_at.isoformat()
                    if s.last_owner_confirmed_at else None
                ),
            }

    data = [
        {
            "id": str(row.id),
            "period_start": row.period_start.isoformat(),
            "status": row.status,
            "created_at": row.created_at.isoformat(),
            "server_count": len(row.server_ids),
            "servers": [server_meta[sid] for sid in row.server_ids if sid in server_meta],
        }
        for row in rows
    ]
    return success_response(data=data, request=request)


@router.post("/server-confirmations/{confirmation_id}/respond")
async def respond_server_confirmation(
    confirmation_id: uuid.UUID,
    body: ServerConfirmationResponse,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Owner responds to a pending server-inventory confirmation.

    Two response shapes:
      * ``kind=confirmed_all`` — owner clicked the single-button CTA
        on the card. Every snapshot server gets its
        ``last_owner_confirmed_at`` bumped to NOW.
      * ``kind=partial`` — owner picked a subset in the management
        drawer. Only those servers are stamped; the row stays
        ``pending`` until the snapshot is fully covered (across one
        or more clicks).
    """
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.server_owner_confirmation_service import (
        ServerOwnerConfirmationService,
    )

    service = ServerOwnerConfirmationService(db)
    try:
        if body.kind == "confirmed_all":
            row = await service.respond_confirm_all(
                confirmation_id=str(confirmation_id),
                actor_person_id=str(user.id),
            )
        else:
            # body.kind == "partial" — Literal type guarantees this.
            if not body.server_ids:
                raise ValidationError("server_ids required for kind='partial'")
            row = await service.respond_per_server(
                confirmation_id=str(confirmation_id),
                actor_person_id=str(user.id),
                confirmed_server_ids=body.server_ids,
            )
    except PermissionError as exc:
        raise ForbiddenError(str(exc)) from exc
    except LookupError as exc:
        raise NotFoundError(str(exc)) from exc
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc

    # Audit. Keeps the timeline readable for the security-auditor
    # surface — every owner self-confirmation is a meaningful
    # governance event, not just a UI click. Log BEFORE commit so the
    # audit row lands in the same transaction; out-of-band failures
    # don't leave us with a stamped server but no audit trail.
    audit = AuditService(db)
    await audit.log(
        trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.server_owner_confirmation.respond",
        status="success",
        resource_type="cmdb_server_owner_confirmation",
        resource_id=str(row.id),
        detail={
            "kind": body.kind,
            "snapshot_size": len(row.server_ids),
            "responded_server_ids": body.server_ids or row.server_ids,
        },
        ip_address=get_real_ip(request),
    )
    await db.commit()

    return success_response(
        data={
            "id": str(row.id),
            "status": row.status,
            "response_kind": row.response_kind,
            "responded_at": row.responded_at.isoformat() if row.responded_at else None,
        },
        request=request,
    )


# ─────────────────────────────────────────────────────────────────────
# 专线管理 (dedicated / leased line register)
# Manually-curated network circuits. Read: can_read_cloud (admin-tier,
# incl. finance/auditor/management — the rows carry cost). Write:
# can_manage_cloud (global_admin / cloud_admin).
# ─────────────────────────────────────────────────────────────────────

def _serialize_dedicated_line(row, labels: dict[str, str]) -> dict:
    item = DedicatedLineResponse.model_validate(row).model_dump()
    modifier = row.last_modified_by_person_id
    if modifier:
        item["last_modified_by_label"] = labels.get(str(modifier))
    return item


@router.get("/dedicated-lines")
async def list_dedicated_lines(
    request: Request,
    search: str | None = None,
    carrier: str | None = None,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()

    rows, total = await dedicated_line_service.list_dedicated_lines(
        db, search=search, carrier=carrier, page=page, page_size=page_size,
    )
    labels = await _resolve_person_labels_for_rows(db, rows)
    data = [_serialize_dedicated_line(r, labels) for r in rows]
    return paginated_response(
        data=data, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/dedicated-lines/{line_id}")
async def get_dedicated_line(
    line_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()

    row = await dedicated_line_service.get_dedicated_line(db, line_id)
    if row is None:
        raise NotFoundError("DedicatedLine", line_id)
    labels = await _resolve_person_labels_for_rows(db, [row])
    return success_response(
        data=_serialize_dedicated_line(row, labels), request=request,
    )


@router.post("/dedicated-lines", status_code=201)
async def create_dedicated_line(
    body: DedicatedLineCreate,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()

    row = await dedicated_line_service.create_dedicated_line(
        db, body, actor_person_id=str(user.id),
    )
    audit = AuditService(db)
    await audit.log(
        trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dedicated_line.create",
        status="success",
        resource_type="cmdb_dedicated_line",
        resource_id=str(row.id),
        detail={
            "location": body.location,
            "carrier": body.carrier,
            "contracting_entity": body.contracting_entity,
            "monthly_cost": body.monthly_cost,
            "annual_cost": body.annual_cost,
            "currency": body.currency,
        },
        ip_address=get_real_ip(request),
    )
    await db.commit()
    await db.refresh(row)
    labels = await _resolve_person_labels_for_rows(db, [row])
    return success_response(
        data=_serialize_dedicated_line(row, labels), request=request,
    )


@router.patch("/dedicated-lines/{line_id}")
async def update_dedicated_line(
    line_id: str,
    body: DedicatedLineUpdate,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()

    row = await dedicated_line_service.update_dedicated_line(
        db, line_id, body, actor_person_id=str(user.id),
    )
    if row is None:
        raise NotFoundError("DedicatedLine", line_id)
    audit = AuditService(db)
    await audit.log(
        trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dedicated_line.update",
        status="success",
        resource_type="cmdb_dedicated_line",
        resource_id=str(row.id),
        # ``mode="json"`` is a no-op on today's all-primitive
        # ``DedicatedLineUpdate``, but ``audit_logs.detail`` is a JSON
        # column with a bare ``json.dumps`` serializer: the first
        # Decimal / datetime / UUID field this schema grows would 500
        # the PATCH at the audit flush, AFTER the row was already
        # mutated. Enforced by ``test_audit_detail_model_dump_mode``.
        detail=body.model_dump(exclude_unset=True, mode="json"),
        ip_address=get_real_ip(request),
    )
    await db.commit()
    await db.refresh(row)
    labels = await _resolve_person_labels_for_rows(db, [row])
    return success_response(
        data=_serialize_dedicated_line(row, labels), request=request,
    )


@router.delete("/dedicated-lines/{line_id}")
async def delete_dedicated_line(
    line_id: str,
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()

    row = await dedicated_line_service.delete_dedicated_line(
        db, line_id, actor_person_id=str(user.id),
    )
    if row is None:
        raise NotFoundError("DedicatedLine", line_id)
    audit = AuditService(db)
    await audit.log(
        trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.dedicated_line.delete",
        status="success",
        resource_type="cmdb_dedicated_line",
        resource_id=str(row.id),
        detail={"location": row.location, "carrier": row.carrier},
        ip_address=get_real_ip(request),
    )
    await db.commit()
    return success_response(data={"id": str(row.id), "deleted": True}, request=request)


# ── Leaver cloud-IAM residue governance ──────────────────────────────
#
# The queue the daily ``cmdb.leaver_iam_residue_sweep`` fills: departed
# people who still hold cloud IAM policies or group memberships. See
# ``leaver_iam_residue.py`` for why the check reads the provider mirror
# rather than Helix's own grant rows.
#
# Remediation deliberately does NOT live here. Deleting a cloud account
# is the existing ``POST /cmdb/iam-users/{id}/cleanup`` chain — one
# implementation, one audit action, one prod-write-lock gate. This
# surface adds the queue, the age clock, and the "known exception"
# disposition; on the next sweep a cleaned-up row resolves itself.


class ResidueDismissRequest(BaseModel):
    reason: str = Field(
        min_length=4, max_length=500,
        description=(
            "Why this residue is acceptable. Required — a dismissal with "
            "no stated reason is indistinguishable from someone clearing "
            "their inbox, and this queue is audit evidence."
        ),
    )


def _residue_row_payload(row: Any, person: Any, now: datetime) -> dict:
    detail = row.detail or {}
    first_seen = row.first_detected_at
    age_days = None
    if first_seen is not None:
        base = first_seen.replace(tzinfo=None) if first_seen.tzinfo else first_seen
        age_days = max(0, (now - base).days)
    days_since_leaving = None
    term = getattr(person, "termination_date", None) if person else None
    if term is not None:
        term_date = term.date() if isinstance(term, datetime) else term
        days_since_leaving = max(0, (now.date() - term_date).days)
    return {
        "id": str(row.id),
        "iam_user_id": str(row.iam_user_id),
        "person_id": str(row.person_id),
        "person_display_name": getattr(person, "display_name", None),
        "person_work_email": getattr(person, "work_email", None),
        "person_employee_no": getattr(person, "employee_no", None),
        "person_employment_status": getattr(person, "employment_status", None),
        "termination_date": term.isoformat() if term is not None else None,
        "days_since_leaving": days_since_leaving,
        "integration_id": str(row.integration_id),
        "integration_name": detail.get("integration_name"),
        "provider": row.provider,
        "user_name": row.user_name,
        "policy_count": row.policy_count,
        "group_count": row.group_count,
        "policy_names": detail.get("policy_names") or [],
        "group_names": detail.get("group_names") or [],
        # ``email`` means the IAM row carried no owner FK and was tied to
        # the person by address — a weaker claim than ``owner``, and the
        # UI says so before anyone authorises an irreversible delete.
        "match_basis": detail.get("match_basis") or "owner",
        # The provider mirror behind this row is older than the IAM sync
        # interval allows for: report it, never act on it.
        "mirror_stale": bool(detail.get("mirror_stale")),
        "status": row.status,
        "first_detected_at": (
            row.first_detected_at.isoformat() if row.first_detected_at else None
        ),
        "last_detected_at": (
            row.last_detected_at.isoformat() if row.last_detected_at else None
        ),
        "open_age_days": age_days,
        "resolution": row.resolution,
        "resolved_at": row.resolved_at.isoformat() if row.resolved_at else None,
        "resolution_note": row.resolution_note,
    }


@router.get("/leaver-iam-residue")
async def list_leaver_iam_residue(
    request: Request,
    status: str = Query(
        default="open", pattern="^(open|dismissed|resolved|all)$",
    ),
    integration_id: str | None = None,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """The governance queue, newest-risk-first.

    Readable by anyone who can read cloud (auditors included — this is
    exactly the evidence an access review asks for); acting on a row
    still requires cloud-manage authority.
    """
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.leaver_iam_residue import summarize_residue
    from app.domains.cmdb.models import CmdbLeaverIamResidue
    from app.identity.person_models import IdentityPerson

    stmt = select(CmdbLeaverIamResidue)
    if status != "all":
        stmt = stmt.where(CmdbLeaverIamResidue.status == status)
    if integration_id:
        stmt = stmt.where(
            CmdbLeaverIamResidue.integration_id == integration_id,
        )
    # Oldest first: the row that has been open longest is the one whose
    # blast radius has had the most time to grow.
    stmt = stmt.order_by(CmdbLeaverIamResidue.first_detected_at.asc())
    rows = (await db.execute(stmt)).scalars().all()

    total = len(rows)
    start = (page - 1) * page_size
    page_rows = rows[start:start + page_size]

    person_ids = {str(r.person_id) for r in page_rows}
    people = {}
    if person_ids:
        people = {
            str(p.id): p for p in (await db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.id.in_([uuid.UUID(i) for i in person_ids]),
                )
            )).scalars().all()
        }

    now = datetime.now(UTC).replace(tzinfo=None)
    items = [
        _residue_row_payload(r, people.get(str(r.person_id)), now)
        for r in page_rows
    ]
    # Hero stats ride in ``meta`` alongside pagination so the page paints
    # its summary from the same request that fills the table — a second
    # round-trip would let the two disagree for a frame.
    return success_response(
        data=items,
        request=request,
        meta={
            "page": page,
            "page_size": page_size,
            "total": total,
            "summary": await summarize_residue(db),
        },
    )


@router.post("/leaver-iam-residue/rescan")
async def rescan_leaver_iam_residue(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Re-run detection now, without waiting for 02:10.

    Cheap and side-effect-light: pure DB reads over the already-synced
    IAM mirror plus the row reconcile. It does NOT re-pull from the
    cloud (that is the IAM capability sync's job) and performs no cloud
    writes, so it is safe to expose as a button.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.leaver_iam_residue import (
        MAX_RESIDUE_ROWS_PER_RUN,
        detect_leaver_iam_residue,
        reconcile_residue_rows,
        summarize_residue,
    )

    findings = await detect_leaver_iam_residue(db)
    if len(findings) > MAX_RESIDUE_ROWS_PER_RUN:
        raise AppException(
            code="RESIDUE_SCAN_TOO_LARGE",
            message=(
                f"Detected {len(findings)} residue rows, past the "
                f"{MAX_RESIDUE_ROWS_PER_RUN} cap. That usually means an "
                "HRIS sync mass-flipped people to 'left'. Verify "
                "identity_persons.employment_status before rescanning."
            ),
            status_code=409,
        )
    result = await reconcile_residue_rows(db, findings)
    return success_response(
        data={
            "detected": result.detected,
            "opened": result.opened,
            "still_open": result.still_open,
            "auto_resolved": result.auto_resolved,
            "summary": await summarize_residue(db),
        },
        request=request,
    )


@router.post("/leaver-iam-residue/{residue_id}/dismiss")
async def dismiss_leaver_iam_residue(
    request: Request,
    residue_id: str,
    body: ResidueDismissRequest,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_browser_session),
):
    """Mark a residue row as a known, accepted exception.

    Does not touch the cloud. The row stops appearing in 待处理 and stops
    contributing to the alert, but stays queryable under 已忽略 with the
    actor and reason attached — a dismissal is a decision someone made,
    and an access review will ask who made it.

    A dismissed row still auto-resolves once the residue actually
    disappears from the mirror, so the 已忽略 list cannot grow forever.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.models import (
        RESIDUE_STATUS_DISMISSED,
        RESIDUE_STATUS_OPEN,
        CmdbLeaverIamResidue,
    )

    row = await db.get(CmdbLeaverIamResidue, uuid.UUID(residue_id))
    if row is None:
        raise NotFoundError("CmdbLeaverIamResidue", residue_id)
    if row.status != RESIDUE_STATUS_OPEN:
        # Idempotent: dismissing an already-dismissed / auto-resolved row
        # is a no-op, not an error — two admins clicking the same row is
        # a normal race in a shared queue.
        return success_response(
            data={"id": str(row.id), "status": row.status},
            request=request,
        )

    row.status = RESIDUE_STATUS_DISMISSED
    row.resolution = None
    row.resolution_note = body.reason
    row.resolved_by_person_id = str(user.id)
    row.resolved_at = datetime.now(UTC).replace(tzinfo=None)

    audit = AuditService(db)
    await audit.log(
        trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.leaver_iam_residue.dismiss",
        status="success",
        resource_type="cmdb_leaver_iam_residue",
        resource_id=str(row.id),
        detail={
            "person_id": str(row.person_id),
            "user_name": row.user_name,
            "provider": row.provider,
            "policy_count": row.policy_count,
            "group_count": row.group_count,
            "reason": body.reason,
        },
        ip_address=get_real_ip(request),
    )
    await db.commit()
    return success_response(
        data={"id": str(row.id), "status": row.status}, request=request,
    )


# ── IAM account classification ───────────────────────────────────────


class IamUserClassifyRequest(BaseModel):
    kind: Literal["human", "service", "unset"] = Field(
        description=(
            "``service`` = a robot that will never map to a person; "
            "``human`` = a colleague Helix has not matched yet (still "
            "needs an owner); ``unset`` clears the classification."
        ),
    )
    note: str | None = Field(
        default=None, max_length=500,
        description="Why. Shown next to the badge and in the audit row.",
    )


@router.post("/iam-users/{iam_user_id}/classify")
async def classify_iam_user(
    request: Request,
    iam_user_id: str,
    body: IamUserClassifyRequest,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_browser_session),
):
    """Say what kind of account this is.

    Compliance surfaces treat every IAM user without an
    ``owner_person_id`` as an orphan, and two different problems land in
    that bucket: a colleague Helix failed to match (act on it) and a
    robot that will never match (stop asking). Prod Tencent has both.
    Without somewhere to record the difference the orphan list can never
    reach zero — and a list that can never reach zero stops being read,
    which is how two departed employees' live AdministratorAccess went
    unnoticed for months.

    Purely a Helix-side statement: no cloud call, nothing revoked,
    nothing granted. Marking a robot as ``service`` takes it out of the
    orphan/residue queues; it does NOT stop syncing it, and its policies
    are still listed.

    ``human`` is deliberately not the same as "assign an owner" — it
    records "this is a person we have not matched yet", which keeps the
    row in the queue as work rather than filing it away.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError(
            "You do not have permission to manage cloud IAM users.",
        )
    from app.domains.cmdb.models import CmdbIamUser

    row = await db.get(CmdbIamUser, uuid.UUID(iam_user_id))
    if row is None or row.deleted_at is not None:
        raise NotFoundError("CmdbIamUser", iam_user_id)

    previous = row.account_kind
    if body.kind == "unset":
        row.account_kind = None
        row.account_kind_note = None
        row.account_kind_set_by_person_id = None
        row.account_kind_set_at = None
    else:
        row.account_kind = body.kind
        row.account_kind_note = (body.note or "").strip() or None
        row.account_kind_set_by_person_id = str(user.id)
        row.account_kind_set_at = datetime.now(UTC).replace(tzinfo=None)

    audit = AuditService(db)
    await audit.log(
        trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.iam_user.classify",
        status="success",
        resource_type="cmdb_iam_user",
        resource_id=str(row.id),
        detail={
            "iam_user_name": row.user_name,
            "integration_id": str(row.integration_id),
            "previous_kind": previous,
            "kind": row.account_kind,
            "note": row.account_kind_note,
        },
        ip_address=get_real_ip(request),
    )
    await db.commit()
    return success_response(
        data=await _enrich_iam_user_response(db, row), request=request,
    )