import re
import uuid
from datetime import date as date_type
from datetime import datetime
from decimal import Decimal
from typing import Any

from sqlalchemy import (
    JSON,
    BigInteger,
    Boolean,
    Date,
    DateTime,
    Float,
    ForeignKey,
    Index,
    Integer,
    Numeric,
    String,
    Text,
    UniqueConstraint,
    text,
)
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.sql import func

# Force-register the identity_persons table into SQLAlchemy MetaData
# whenever this module is loaded. Several FKs below use string references
# like ``ForeignKey("identity_persons.id", ...)`` which SQLAlchemy resolves
# lazily at mapper-configure time (first ORM op). In the API process the
# target is registered transitively via ``auth.models``, but the worker's
# boot path never imports auth, so ``cmdb.resource_sync`` could hit a
# fresh process and die with "Foreign key associated with column
# 'cmdb_sync_jobs.triggered_by_user_id' could not find table
# 'identity_persons'" (observed in prod 2026-04-24). Pinning the import
# here documents the cross-domain dependency where the FK is declared.
import app.identity.person_models  # noqa: F401, E402
from app.core.database import Base
from app.core.sql_types import GUID as Uuid

# ─── Cloud scope / grant lifecycle constants ────────────────────────
# Keep these as module-level constants so both the service, sweeps,
# routers, and offboarding code share one vocabulary.

SCOPE_TYPE_AWS_ACCOUNT = "aws_account"          # AWS Org member account
SCOPE_TYPE_AZURE_SUBSCRIPTION = "azure_subscription"
SCOPE_TYPE_GCP_PROJECT = "gcp_project"
SCOPE_TYPE_ALIYUN_ACCOUNT = "aliyun_account"    # Alicloud RD member account
# Single-tenant IAM-mode clouds (AWS SG, Tencent) — the "tenant" and
# "scope" collapse to one row. Keeping scope_type explicit makes the
# request UI a one-branch decision: "this cloud has sub-accounts" vs
# "this cloud has IAM users only".
SCOPE_TYPE_STANDALONE_IAM = "standalone_iam"
# Sub-tenant scope: Azure Resource Group inside a subscription. NOT a
# unit of cloud-lifecycle access (people don't "request access" to an
# RG via the Helix self-service grant flow); it's the cost-binding
# scope for the Azure OpenAI auto-grant. Each approved AI API key
# request mints a fresh RG row whose ``parent_scope_id`` points at
# the owning ``azure_subscription`` scope.
#
# ``external_id`` holds an RG NAME, not a subscription GUID — so these
# rows must be kept out of every subscription-addressed provider call.
# The filter is ``TENANT_LEVEL_SCOPE_TYPES`` below, applied by the
# requestable-scopes query (``lifecycle_router.list_scopes``), the
# grants / roles / billing / cost sync loops, and the service-layer
# grant gate ``_refuse_ungrantable_scope_type``. (An earlier version of
# this comment named ``CmdbScopeService.list_requestable``, which has
# never existed — the missing picker filter is exactly how an AOAI RG
# row reached the employee request list.)
SCOPE_TYPE_AZURE_RESOURCE_GROUP = "azure_resource_group"
# GCP billing account (``billingAccounts/XXXX-XXXX-XXXX``). A grantable
# scope of its OWN — GCP billing roles (roles/billing.viewer etc.) live
# at the billing-account level, NOT the project, so "I want GCP-console
# billing visibility" is satisfied by a grant on THIS scope, not on a
# gcp_project. Grants route to the Cloud Billing API's
# ``billingAccounts:setIamPolicy`` (see GcpCloudDriver). Discovered by
# scope sync via ``cloudbilling.billingAccounts.list``; not auto-owned
# (admin assigns a billing/finance owner to open it for requests).
SCOPE_TYPE_GCP_BILLING_ACCOUNT = "gcp_billing_account"
# A cost-allocation unit imported from a DECOMMISSIONED tenant, kept so
# historical bills can be attributed. NOT a cloud account and NOT an
# access scope: nobody can request access to one, no driver can address
# one, and no owner governance applies.
#
# Prod 2026-08-05: a one-off script created three "(legacy billing)"
# integrations (KSCloud / UCloud / AWS China, all ``enabled=0``,
# ``capabilities=[]``) plus 37 scope rows in a single second, typed
# ``standalone_iam`` — so they rendered as 37 live cloud subscriptions in
# 云订阅 with owners attached. They are not sub-accounts: 金山云 is
# project-based, and the names are PingCAP cost centres
# (``C0801-RDG-TiDB_Cloud``, ``默认项目``, ``Innovation-Project``). The
# bills were never imported, so they sat there as pure noise on the
# access-control surface.
#
# Typing them keeps the rows (the bills are still meant to be imported)
# while removing them from every access surface by construction: this
# type is absent from ``TENANT_LEVEL_SCOPE_TYPES``, which the requestable
# picker, the grants / roles / billing / cost sync fan-outs and the
# grant gate all filter on.
SCOPE_TYPE_LEGACY_BILLING_UNIT = "legacy_billing_unit"

# Tenant-level scope types — the units of cloud-lifecycle access AND
# the only scope rows whose ``external_id`` is a provider-addressable
# tenant id (subscription GUID / account id / project id). Used by the
# requestable-scopes filter so RG-scope rows (which exist purely for
# cost binding) don't pollute the access-request UI, and by every
# per-scope driver fan-out (grants / roles / billing_daily /
# cost_facts) so an RG name never lands in a provider URL path.
TENANT_LEVEL_SCOPE_TYPES: frozenset[str] = frozenset({
    SCOPE_TYPE_AWS_ACCOUNT,
    SCOPE_TYPE_AZURE_SUBSCRIPTION,
    SCOPE_TYPE_GCP_PROJECT,
    SCOPE_TYPE_ALIYUN_ACCOUNT,
    SCOPE_TYPE_STANDALONE_IAM,
    SCOPE_TYPE_GCP_BILLING_ACCOUNT,
})

SCOPE_STATUS_ACTIVE = "active"
SCOPE_STATUS_SUSPENDED = "suspended"
SCOPE_STATUS_CLOSED = "closed"


def owner_governed_scope_criteria() -> list[Any]:
    """WHERE criteria for "owning this scope is a LIVE, TRANSFERABLE duty".

    Use this — never a hand-rolled ``deleted_at.is_(None)`` — on every
    surface that asks a human to *do* something because they own a
    scope: offboarding blockers, the offboarding preview, mover /
    role-change holdings, the manager team view, leaver residue.

    Three predicates, each earning its place:

    * ``deleted_at IS NULL`` — the row exists.
    * ``status != closed`` / ``decommissioned_at IS NULL`` —
      decommission is irreversible, so a closed scope's
      ``owner_person_id`` is bookkeeping, not a duty. Nobody can
      "hand over" an account that no longer exists. This mirrors the
      sibling IAM-user scanner's ``revoked_at IS NULL`` and the
      reason its comment gives: asking a new manager about it is noise.
    * ``scope_type IN TENANT_LEVEL_SCOPE_TYPES`` — a
      ``legacy_billing_unit`` is a cost-allocation row from a dead
      tenant (migration 466: "no owner governance applies") and an
      ``azure_resource_group`` is a cost-binding anchor minted per
      approved AOAI API key, whose real handover path is
      ``transfer_apikey_assignment_owner``. Neither is an account
      anyone can be handed.

    **Why a shared helper and not four copies of the WHERE clause.**
    These queries live in four different modules and had drifted into
    three different answers about the same person: on 2026-09-07 the
    admin console showed 7 scopes for an owner, the employee page 12,
    the leaver-residue reconciler applied the status filter but not the
    type filter, and the offboarding scanner applied neither — so it
    raised `cloud_scope_owner` blockers on rows that are not accounts
    (a ``closed`` resource group, a ``legacy_billing_unit``), which the
    T+0 force-transfer path then "resolved" by handing a successor the
    ownership of a cost-allocation row. (Those blockers were NOT stuck:
    a first revision of this docstring said nobody could resolve them,
    which was inferred rather than checked — prod has zero unresolved
    blockers of any type. The defect is meaningless churn, not a
    wedged process.) Copies of a predicate drift silently because
    nothing fails when they disagree; one function cannot.
    """
    return [
        CmdbCloudScope.deleted_at.is_(None),
        CmdbCloudScope.status != SCOPE_STATUS_CLOSED,
        CmdbCloudScope.decommissioned_at.is_(None),
        CmdbCloudScope.scope_type.in_(TENANT_LEVEL_SCOPE_TYPES),
    ]


def employee_directory_scope_criteria() -> list[Any]:
    """WHERE criteria for "show this to an employee as a cloud account".

    The discovery half of the same coin: what 我的云订阅's requestable
    directory and the admin 云子账号/订阅 list must agree on. Same
    ungrantable-type exclusion as ``owner_governed_scope_criteria``,
    plus two differences that are deliberate, not drift:

    * ``status == active`` (not merely "not closed") — a suspended
      account is not something to offer for a fresh access request.
    * ``standalone_iam`` excluded — Tencent / UCloud / AWS-Singapore
      hand out an IAM account or API key rather than a role on a shared
      tenant, so they live on the IAM-users page. That is a page-level
      routing decision, NOT an ungrantability one, which is exactly why
      it is subtracted here and kept in the owner-governance set above.
    """
    return [
        CmdbCloudScope.deleted_at.is_(None),
        CmdbCloudScope.status == SCOPE_STATUS_ACTIVE,
        CmdbCloudScope.scope_type.in_(
            TENANT_LEVEL_SCOPE_TYPES - {SCOPE_TYPE_STANDALONE_IAM},
        ),
    ]

GRANT_STATUS_PENDING = "pending"    # awaiting scope owner approval
GRANT_STATUS_ACTIVE = "active"
GRANT_STATUS_EXPIRED = "expired"
GRANT_STATUS_REVOKED = "revoked"
GRANT_STATUS_DISABLED = "disabled"  # auto-disabled by inactivity sweep

# Terminal-state grants whose row may be resurrected by a fresh
# request (see ``CloudLifecycleService.submit_access_request`` /
# ``owner_add_member``). The unique key
# ``uq_cmdb_cloud_grants_scope_person_role`` does not partition by
# status, so a re-submit MUST reuse the existing row in-place
# instead of inserting a duplicate.
GRANT_STATUSES_RESURRECTABLE: frozenset[str] = frozenset({
    GRANT_STATUS_REVOKED,
    GRANT_STATUS_EXPIRED,
    GRANT_STATUS_DISABLED,
})

# Closure-of-access statuses counted as "回收". The cloud overview's
# ``recovered.*`` counters (admin_overview_service) and the insights
# impact scorecard's cloud reclaim counters MUST both use this set —
# gated on a stamped ``revoked_at`` — or the two surfaces drift
# (2026-06-12: the scorecard counted only ``revoked`` and reported 980
# while the overview reported 1131; the 151 gap was ``disabled`` rows
# from decommissions and inactivity sweeps). Same membership as
# ``GRANT_STATUSES_RESURRECTABLE`` today, but the two sets answer
# different questions — keep them separate.
GRANT_STATUSES_RECLAIMED: tuple[str, ...] = (
    GRANT_STATUS_REVOKED,
    GRANT_STATUS_EXPIRED,
    GRANT_STATUS_DISABLED,
)

# Scope ``environment`` values that are excluded from every Helix
# self-service grant flow (request + owner-add). Production cloud
# access is owned by the R&D security team (Kidd.li) outside Helix
# self-service. The block applies to every caller — admins who
# legitimately need to provision prod use the security team's
# intake, not a Helix endpoint. Centralised so the router filter
# (``list_scopes?requestable=true``) and the service-level guards
# stay in lock-step if the prod-environment vocabulary ever grows.
PROD_ENVIRONMENTS: frozenset[str] = frozenset({"prod"})

# Default ``environment`` for a scope nobody has classified. Matches the
# column's server_default; named so the discovery-time classifier and the
# model stay in lock-step.
SCOPE_ENVIRONMENT_DEFAULT = "dev"


# How a server declares which cloud subscription it belongs to. Every sync
# driver stamps exactly ONE of these onto ``CmdbServer.raw_data``, and the
# value is the same string the owning ``CmdbCloudScope`` carries in
# ``external_id`` — so this is the provider's own account membership, not an
# inference from names or tags. ``CmdbServer`` has no scope FK; this map plus
# billing attribution are the only links that exist.
#
# Keyed by ``scope_type`` for the scope-side SQL predicate
# (``_scope_inventory_clause``); ``SCOPE_ACCOUNT_RAW_KEYS`` is the derived
# plain-key tuple for the server side (dict lookups and JSON equality in
# ``visibility_clause``). Derived, never restated: this existing once and
# being wired on only one side is exactly how AWS/Azure servers ended up with
# no resolvable subscription for months (2026-08-18). Adding a provider here
# wires the resolver, the inherited-owner reader AND self-service visibility
# at once. UCloud's driver stamps nothing, so its servers rely on billing.
SCOPE_RAW_PATHS: dict[str, str] = {
    "aws_account": "$._aws_account_id",
    "aliyun_account": "$.__sub_account_id",
    "azure_subscription": "$._subscription_id",
    "gcp_project": "$._project_id",
}

SCOPE_ACCOUNT_RAW_KEYS: tuple[str, ...] = tuple(
    path.removeprefix("$.") for path in SCOPE_RAW_PATHS.values()
)

# "prod" as a whole token inside a cloud account NAME. Deliberately NOT a
# plain ``"prod" in name`` substring test: every ``*-nonprod-*`` account
# CONTAINS "prod", so the naive check classifies the entire non-production
# estate as production. On 2026-08-06 that would have mislabelled 8 of the
# 17 newly-onboarded sys9/infra accounts (``sys9-nonprod-eu-stg``,
# ``infra-nonprod-gateway``, …) as prod — locking them behind the prod
# write lock and exempting them from expiry sweeps.
#
# Three guards, each earning its place:
#   * ``(?<![a-z])``      — a letter before "prod" means it's part of a
#                           longer word: ``nonprod``, ``reproduction``.
#   * ``(?<!non-)`` /
#     ``(?<!non_)``       — the separator forms. ``(?<![a-z])`` alone
#                           HAPPILY matches ``non-prod`` (the char before
#                           "prod" is "-", not a letter), which is the
#                           exact bug this comment exists to prevent —
#                           caught by the parametrised test, not by eye.
#   * ``(?![a-z])``       — after the token; keeps ``production`` a match
#                           while rejecting ``prodigy``.
_PROD_NAME_TOKEN = re.compile(
    r"(?<![a-z])(?<!non-)(?<!non_)prod(uction)?(?![a-z])",
)


def classify_scope_environment(display_name: str | None) -> str:
    """Best-effort ``environment`` for a NEWLY discovered cloud scope.

    Returns ``"prod"`` when the provider-reported account name carries a
    standalone ``prod`` / ``production`` token, else
    ``SCOPE_ENVIRONMENT_DEFAULT``.

    Why name-based: AWS Organizations / Azure / GCP expose no environment
    attribute, and the account name is the only signal present at
    discovery time. Before this existed EVERY synced scope landed ``dev``
    and stayed there until a human noticed — production account
    ``417183877957`` (``dbaas-prod-essential-v2``) sat mislabelled as 开发
    from first sync until an employee reported it.

    Applied ONLY on the INSERT branch of the scope upsert. A human's
    later correction (``update_scope_config``) must never be reverted by
    the next sync, and names are a heuristic — ``pingcap-dbaas`` is
    production but says nothing about it, so the reverse inference is
    unsafe and deliberately absent.
    """
    if not display_name:
        return SCOPE_ENVIRONMENT_DEFAULT
    if _PROD_NAME_TOKEN.search(display_name.lower()):
        return "prod"
    return SCOPE_ENVIRONMENT_DEFAULT

# ── Grant TTL policy ───────────────────────────────────────────────
#
# Before 2026-07-29 the request form accepted any TTL in 1..365 with no
# tiering, and the requester's own number won: the TTL resolution chain
# in ``_dispatch_grant`` places ``detail.requested_ttl_days`` AHEAD of
# ``scope.default_grant_ttl_days``, so a scope configured for 90 days
# handed out 365 whenever someone typed 365. Production held grants at
# 10 / 30 / 90 / 180 / 365 days side by side on the same account, which
# makes "90-day access review" an unenforceable claim.
#
# Every TTL — the initial request AND every renewal — is now bounded by
# a single hard cap, and the scope owner is a sufficient approver for
# anything under it:
#
#   * ``GRANT_TTL_DEFAULT_DAYS`` is what the forms pre-fill and what a
#     request with no explicit TTL resolves to;
#   * above ``GRANT_TTL_HARD_CAP_DAYS`` nothing is grantable at all. Need
#     more than that and the answer is a permanent role, not a 2-year
#     grant nobody will ever review.
#
# 2026-08-26: the middle tier ("above 90 days a cloud admin must also
# sign") was REMOVED at the user's direction — the scope owner is the
# authority on their own subscription, and a second signature between 91
# and the 180-day cap bought no governance the cap does not already buy.
# See DECISIONS.md 2026-08-26.
GRANT_TTL_DEFAULT_DAYS = 90
GRANT_TTL_HARD_CAP_DAYS = 180

GRANT_NOTIFY_T14 = "t14"
GRANT_NOTIFY_T7 = "t7"
GRANT_NOTIFY_T1 = "t1"
GRANT_NOTIFY_INACTIVE_OWNER = "inactive_owner"
GRANT_NOTIFY_INACTIVE_IT = "inactive_it"
# Stamped on prod scopes whose grant has passed expires_at — the
# expiry sweeper holds them active (prod is opted out of auto-revoke)
# but pings the scope owner exactly once instead of every hour.
GRANT_NOTIFY_PROD_OVERDUE = "prod_overdue"

# ── Renewal-request state machine ──────────────────────────────────
#
# Renewal used to be a self-service one-click action: the holder pressed
# 「一键续期」 (or the button on the expiry reminder card) and the grant's
# clock reset immediately, with no second party involved. That made
# "expiry recycling" unenforceable in the other direction too — an
# already-recycled grant could be revived by its former holder alone, at
# any time, with no age limit, because ``_renew_grant_locked`` accepted
# any row in status ``expired``. 839 recycled grants across 135 people
# were self-revivable that way on 2026-07-29.
#
# Renewal is now a request that the scope owner approves, and revival of
# a recycled grant goes through the same gate.
RENEWAL_STATUS_PENDING_OWNER = "pending_owner"
# RETIRED 2026-08-26 — the owner's signature is now final at every TTL,
# so nothing writes this status any more. The constant stays because it
# is a persisted enum value: history rows and the status→colour /
# status→label maps that read them must keep resolving it, and a rename
# would cost every reader forever to save one line here. The decide path
# still accepts a row found in this state (owner already signed) and
# finalizes it, so a leftover row cannot become undecidable — production
# had none when the second signature was removed.
RENEWAL_STATUS_PENDING_ADMIN = "pending_admin"
RENEWAL_STATUS_APPROVED = "approved"
RENEWAL_STATUS_REJECTED = "rejected"
# No decision inside ``RENEWAL_REQUEST_LAPSE_DAYS`` — closed by cron so a
# forgotten request cannot sit open forever and be approved months later
# against a grant whose context has changed.
RENEWAL_STATUS_LAPSED = "lapsed"
# Withdrawn by the requester, or auto-closed because the underlying grant
# was hard-revoked / the scope was decommissioned while the request was
# still open.
RENEWAL_STATUS_CANCELLED = "cancelled"

RENEWAL_STATUSES_OPEN: tuple[str, ...] = (
    RENEWAL_STATUS_PENDING_OWNER,
    RENEWAL_STATUS_PENDING_ADMIN,
)
RENEWAL_STATUSES_TERMINAL: tuple[str, ...] = (
    RENEWAL_STATUS_APPROVED,
    RENEWAL_STATUS_REJECTED,
    RENEWAL_STATUS_LAPSED,
    RENEWAL_STATUS_CANCELLED,
)

# How long an undecided renewal request stays actionable. Matches the
# T-14 first-reminder stage: the request is raised 14 days before expiry,
# so a 14-day window means the owner has the entire runway to decide and
# a request never outlives the expiry event that motivated it by more
# than a couple of weeks.
RENEWAL_REQUEST_LAPSE_DAYS = 14

# How long after recycling a grant may still be REVIVED by renewal
# rather than a fresh access request.
#
# ``_renew_grant_locked``'s docstring claimed "expired within 30 days can
# still be revived" since it was written, but no code ever enforced it —
# the only 30 in the function was the action-token TTL. Any expired row
# was revivable forever, so on 2026-07-29 the oldest self-revivable grant
# had been recycled 32 days earlier and nothing would ever have aged it
# out.
#
# The window matters even now that revival needs owner approval: past it,
# the requester should go through a fresh access request, which re-runs
# role governance (elevated/restricted tiers), the driver role-shape
# preflight, and the scope's current requestability rules. A months-old
# grant row is a stale description of what the person needs.
RENEWAL_REVIVAL_MAX_AGE_DAYS = 30

GRANT_REVOKE_REASON_USER = "user_released"
GRANT_REVOKE_REASON_EXPIRED = "expired"
GRANT_REVOKE_REASON_INACTIVE = "inactive"
GRANT_REVOKE_REASON_OFFBOARDING = "offboarding"
GRANT_REVOKE_REASON_OWNER = "owner_removed"
GRANT_REVOKE_REASON_ADMIN = "admin_removed"
# Stamped by the grants-sync mark-and-sweep when a previously
# discovered grant no longer appears in the cloud's authoritative
# response. Distinct from "inactive" / "owner_removed" because the
# action originated cloud-side (admin removed the IAM user / role
# binding directly), not from any Helix workflow. Avoids the false
# stale-grant inactivity notifications that bug bit at PR #170 prep.
GRANT_REVOKE_REASON_DRIFT_REMOVED = "drift_removed"
# Stamped when an admin disowns (cloud-deletes) an IAM user via
# ``POST /cmdb/iam-users/{id}/cleanup``. The cleanup endpoint cascades
# to all active grants attributed to that IAM user — discovered grants
# (matched via the ``iam-user:{name}::{role}`` external_ref pattern)
# AND helix-issued grants for the same person on the same scope (the
# IAM user was the auth handle that enacted the helix grant). Without
# this cascade, the user's "成员 / 生效 grant" hero-stat continued to
# show ghost grants for an IAM user that no longer existed cloud-side
# until the next ``sync_grants_for_integration`` drift sweep — and
# that sweep skips ``source='helix'`` rows by design, so helix grants
# would NEVER auto-clear without this in-band cascade. User report:
# 2026-05-05 / AWS SG (Singapore) "37 个生效 grant 删了 IAM 还在".
GRANT_REVOKE_REASON_IAM_DISOWNED = "iam_user_disowned"
# Stamped when an admin sweeps a cloud-side "orphan" assignment — an
# Identity Center / IAM principal that the cloud console still binds
# to a permission-set on a sub-account, but the principal itself is
# gone (deleted from the SCIM-synced directory). Helix never had a
# Helix-side ``person_id`` for these because the principal was
# already missing the first time ``list_scope_grants`` saw it.
# History rows carry ``person_id IS NULL`` and stash the AWS-side
# principal UUID + best-effort recovered email in
# ``external_principal_id`` / ``external_principal_email`` so the
# admin "已结束" tab can still attribute the swept assignment.
# Audit: ``cloud.grant.orphan_principal_swept``.
GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL = "orphan_principal_swept"
# Stamped by the offboarding leaver-sweep: when a person leaves, we
# resolve their still-live Identity Center principal and delete EVERY
# account assignment they hold — including ones Helix never tracked as a
# grant row (created directly in the AWS console / another tool). This
# runs BEFORE SCIM deprovision deletes the directory user, so those
# untracked assignments never become blank-name orphans (the class the
# ``orphan_principal_swept`` path cleans up after the fact). Unlike the
# orphan sweep, the principal is still attributable here: history rows
# carry the real ``person_id`` plus ``external_principal_id`` /
# ``external_principal_email``. Audit: ``cloud.grant.leaver_principal_swept``.
GRANT_REVOKE_REASON_LEAVER_PRINCIPAL_SWEPT = "leaver_principal_swept"
# Stamped when a scope is closed (via the explicit decommission
# endpoint OR the periodic scope_sync sweep that closes scopes the
# cloud no longer returns) AND there are still active / pending
# grants on that scope. The grants-sync drift sweep is restricted to
# ``synced_scope_ids`` for safety, so without this cascade closed
# scopes accumulate ``status='active'`` zombie grants that never
# get cleaned up (production incident 2026-05-28: 29 active UCloud
# grants stranded on a closed scope 24 days post-close because the
# scope_sync sweep flipped ``status='closed'`` without cascading to
# its grants). Same severity class as the explicit-decommission
# cascade at ``decommission_scope`` (which already uses
# ``GRANT_STATUS_DISABLED``); we use the same disabled status here
# for symmetry and a distinct revoke_reason so compliance audits
# can tell the two close paths apart.
GRANT_REVOKE_REASON_SCOPE_CLOSED = "scope_closed"

# ``CmdbProviderAccount`` was deleted in migration 045; cloud tenants
# are now rows in ``integrations(kind='cloud')``. Every resource table
# below references ``integrations.id`` via the renamed
# ``integration_id`` column.


class CmdbSyncJob(Base):
    __tablename__ = "cmdb_sync_jobs"
    __table_args__ = (
        # Backs the sync-to-sync baseline lookup in ``_safe_sweep``:
        # ``WHERE integration_id = ? AND status IN (success,
        # partial_failure) ORDER BY finished_at DESC LIMIT 20``. Without
        # this composite, the sort happens on a per-integration scan of
        # all historical rows — fine at <1k rows, slow at >50k. See
        # migration 165 for rationale.
        Index(
            "ix_cmdb_sync_jobs_integration_status_finished",
            "integration_id", "status", "finished_at",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    triggered_by: Mapped[str] = mapped_column(String(16), nullable=False)
    triggered_by_user_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    resource_types: Mapped[list[str]] = mapped_column(JSON, nullable=False)
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="queued")
    capability_results: Mapped[dict | None] = mapped_column(JSON)
    total_upserted: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    total_deleted: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    total_errors: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    error_message: Mapped[str | None] = mapped_column(Text)
    queued_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class CmdbResourceType(Base):
    """Runtime catalog for every resource kind observed by Helix.

    Provider inventory APIs are intentionally open-ended: a newly supported
    cloud service must be visible as a generic resource without waiting for a
    new enum, migration, or frontend route.  A dedicated enrichment adapter
    may later add friendly fields, but this row is the durable type identity.
    """

    __tablename__ = "cmdb_resource_types"
    __table_args__ = (
        UniqueConstraint("provider", "resource_type", name="uq_cmdb_resource_type_provider_kind"),
        Index("ix_cmdb_resource_types_category", "category", "active"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    provider: Mapped[str] = mapped_column(String(64), nullable=False)
    resource_type: Mapped[str] = mapped_column(String(192), nullable=False)
    service: Mapped[str] = mapped_column(String(96), nullable=False)
    category: Mapped[str] = mapped_column(String(64), nullable=False)
    display_name: Mapped[str] = mapped_column(String(192), nullable=False)
    icon_key: Mapped[str | None] = mapped_column(String(64))
    sensitivity: Mapped[str] = mapped_column(
        String(16), nullable=False, default="internal", server_default="internal",
    )
    schema_version: Mapped[int] = mapped_column(
        Integer, nullable=False, default=1, server_default="1",
    )
    display_schema: Mapped[dict | None] = mapped_column(JSON)
    discovered_dynamically: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="0",
    )
    active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="1",
    )
    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    last_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )


class CmdbCloudResource(Base):
    """Provider-neutral read model for the cloud resource directory.

    Existing specialised CMDB tables remain authoritative for workflows such
    as server efficiency and DNS lifecycle.  This index supplies one stable,
    searchable identity for inventory, governance, reporting, and AI.  Generic
    provider-discovery rows use the same contract and therefore appear without
    a new page or schema migration.
    """

    __tablename__ = "cmdb_cloud_resources"
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "resource_identity_hash",
            name="uq_cmdb_cloud_resource_identity_hash",
        ),
        UniqueConstraint("source_table", "source_id", name="uq_cmdb_cloud_resource_source"),
        Index("ix_cmdb_cloud_resources_type_status", "resource_type", "lifecycle_status"),
        Index("ix_cmdb_cloud_resources_provider_scope", "provider", "scope_external_id"),
        Index(
            "ix_cmdb_cloud_resources_integration_scope_deleted",
            "integration_id", "scope_external_id", "deleted_at",
        ),
        Index("ix_cmdb_cloud_resources_owner", "owner_person_id", "deleted_at"),
        Index("ix_cmdb_cloud_resources_seen", "last_seen_at", "deleted_at"),
        Index("ix_cmdb_cloud_resources_category_region", "category", "region"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False,
    )
    resource_type_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("cmdb_resource_types.id"), nullable=False,
    )
    provider: Mapped[str] = mapped_column(String(64), nullable=False)
    service: Mapped[str] = mapped_column(String(96), nullable=False)
    category: Mapped[str] = mapped_column(String(64), nullable=False)
    resource_type: Mapped[str] = mapped_column(String(192), nullable=False)
    provider_resource_id: Mapped[str] = mapped_column(Text, nullable=False)
    resource_identity_hash: Mapped[str] = mapped_column(String(64), nullable=False)
    name: Mapped[str] = mapped_column(String(512), nullable=False)
    scope_external_id: Mapped[str | None] = mapped_column(String(256))
    region: Mapped[str | None] = mapped_column(String(128))
    zone: Mapped[str | None] = mapped_column(String(128))
    lifecycle_status: Mapped[str | None] = mapped_column(String(64))
    environment: Mapped[str | None] = mapped_column(String(32))
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    sensitivity: Mapped[str] = mapped_column(
        String(16), nullable=False, default="internal", server_default="internal",
    )
    tags: Mapped[dict | None] = mapped_column(JSON)
    normalized_metadata: Mapped[dict | None] = mapped_column(JSON)
    raw_metadata: Mapped[dict | None] = mapped_column(JSON)
    content_hash: Mapped[str] = mapped_column(String(64), nullable=False)
    source_table: Mapped[str] = mapped_column(String(96), nullable=False)
    source_id: Mapped[str] = mapped_column(String(36), nullable=False)
    source_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    last_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    last_changed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    missing_since: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(),
    )


class CmdbCloudResourceChange(Base):
    """Append-only, bounded resource change evidence for trend/report use."""

    __tablename__ = "cmdb_cloud_resource_changes"
    __table_args__ = (
        Index("ix_cmdb_resource_changes_resource_at", "resource_id", "observed_at"),
        Index("ix_cmdb_resource_changes_type_at", "change_type", "observed_at"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    resource_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("cmdb_cloud_resources.id", ondelete="CASCADE"), nullable=False,
    )
    change_type: Mapped[str] = mapped_column(String(24), nullable=False)
    changed_fields: Mapped[list | None] = mapped_column(JSON)
    previous_hash: Mapped[str | None] = mapped_column(String(64))
    current_hash: Mapped[str | None] = mapped_column(String(64))
    snapshot: Mapped[dict | None] = mapped_column(JSON)
    source_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    observed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )


class CmdbResourceGovernance(Base):
    """Helix-local stewardship metadata; never written back to a provider."""

    __tablename__ = "cmdb_resource_governance"
    __table_args__ = (
        UniqueConstraint("resource_id", name="uq_cmdb_resource_governance_resource"),
        Index("ix_cmdb_resource_governance_risk", "risk_level", "governance_status"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    resource_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("cmdb_cloud_resources.id", ondelete="CASCADE"), nullable=False,
    )
    business_owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    technical_owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    governance_status: Mapped[str] = mapped_column(
        String(24), nullable=False, default="unreviewed", server_default="unreviewed",
    )
    criticality: Mapped[str | None] = mapped_column(String(16))
    data_classification: Mapped[str | None] = mapped_column(String(24))
    risk_level: Mapped[str | None] = mapped_column(String(16))
    lifecycle_stage: Mapped[str | None] = mapped_column(String(24))
    compliance_status: Mapped[str | None] = mapped_column(String(24))
    exception_reason: Mapped[str | None] = mapped_column(Text)
    exception_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    notes: Mapped[str | None] = mapped_column(Text)
    custom_fields: Mapped[dict | None] = mapped_column(JSON)
    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1, server_default="1")
    updated_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(),
    )


class CmdbResourceGovernanceRule(Base):
    """Versioned rule configuration for deterministic resource checks."""

    __tablename__ = "cmdb_resource_governance_rules"
    __table_args__ = (
        UniqueConstraint("rule_key", name="uq_cmdb_resource_governance_rule_key"),
        Index("ix_cmdb_resource_governance_rules_enabled", "enabled", "severity"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    rule_key: Mapped[str] = mapped_column(String(96), nullable=False)
    name: Mapped[str] = mapped_column(String(192), nullable=False)
    description: Mapped[str | None] = mapped_column(Text)
    category: Mapped[str] = mapped_column(String(64), nullable=False)
    severity: Mapped[str] = mapped_column(String(16), nullable=False)
    evaluator: Mapped[str] = mapped_column(String(64), nullable=False)
    applies_to: Mapped[dict | None] = mapped_column(JSON)
    parameters: Mapped[dict | None] = mapped_column(JSON)
    recommendation: Mapped[str | None] = mapped_column(Text)
    enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="1",
    )
    built_in: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="0",
    )
    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1, server_default="1")
    created_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    updated_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(),
    )


class CmdbResourceFinding(Base):
    """Current and historical governance findings keyed by rule/resource."""

    __tablename__ = "cmdb_resource_findings"
    __table_args__ = (
        UniqueConstraint("fingerprint", name="uq_cmdb_resource_finding_fingerprint"),
        Index("ix_cmdb_resource_findings_status_severity", "status", "severity"),
        Index("ix_cmdb_resource_findings_resource_status", "resource_id", "status"),
        Index("ix_cmdb_resource_findings_rule_seen", "rule_id", "last_seen_at"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    resource_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("cmdb_cloud_resources.id", ondelete="CASCADE"), nullable=False,
    )
    rule_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("cmdb_resource_governance_rules.id", ondelete="CASCADE"), nullable=False,
    )
    fingerprint: Mapped[str] = mapped_column(String(64), nullable=False)
    severity: Mapped[str] = mapped_column(String(16), nullable=False)
    status: Mapped[str] = mapped_column(
        String(24), nullable=False, default="open", server_default="open",
    )
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    description: Mapped[str | None] = mapped_column(Text)
    evidence: Mapped[dict | None] = mapped_column(JSON)
    recommendation: Mapped[str | None] = mapped_column(Text)
    assigned_to_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    last_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    acknowledged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    acknowledged_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    resolution_note: Mapped[str | None] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(),
    )


class CmdbCloudResourceDailyRollup(Base):
    """Indefinite, non-resource-level trend facts after detail retention expires."""

    __tablename__ = "cmdb_cloud_resource_daily_rollups"
    __table_args__ = (
        UniqueConstraint("snapshot_date", "dimension_key", name="uq_cmdb_resource_daily_dimension"),
        Index("ix_cmdb_resource_daily_provider_date", "provider", "snapshot_date"),
        Index("ix_cmdb_resource_daily_category_date", "category", "snapshot_date"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    snapshot_date: Mapped[date_type] = mapped_column(Date, nullable=False)
    dimension_key: Mapped[str] = mapped_column(String(64), nullable=False)
    provider: Mapped[str | None] = mapped_column(String(64))
    category: Mapped[str | None] = mapped_column(String(64))
    resource_type: Mapped[str | None] = mapped_column(String(192))
    scope_external_id: Mapped[str | None] = mapped_column(String(256))
    total_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    active_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    missing_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    unowned_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    restricted_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    high_risk_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    open_finding_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(),
    )


class CmdbResourceReportSubscription(Base):
    """Per-person delivery preferences for the permission-scoped directory report."""

    __tablename__ = "cmdb_resource_report_subscriptions"
    __table_args__ = (
        UniqueConstraint("person_id", name="uq_cmdb_resource_report_subscription_person"),
        Index("ix_cmdb_resource_report_due", "frequency", "next_run_at"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    person_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="CASCADE"), nullable=False,
    )
    frequency: Mapped[str] = mapped_column(
        String(16), nullable=False, default="off", server_default="off",
    )
    channels: Mapped[list | None] = mapped_column(JSON)
    timezone: Mapped[str] = mapped_column(
        String(64), nullable=False, default="Asia/Shanghai", server_default="Asia/Shanghai",
    )
    delivery_hour: Mapped[int] = mapped_column(
        Integer, nullable=False, default=9, server_default="9",
    )
    weekly_day: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    monthly_day: Mapped[int] = mapped_column(
        Integer, nullable=False, default=1, server_default="1",
    )
    include_ai: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="0",
    )
    scope_config: Mapped[dict | None] = mapped_column(JSON)
    next_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1, server_default="1")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(),
    )


class CmdbResourceReportRun(Base):
    """Frozen, recipient-specific report evidence created after a fresh permission check."""

    __tablename__ = "cmdb_resource_report_runs"
    __table_args__ = (
        Index("ix_cmdb_resource_report_runs_person_at", "recipient_person_id", "created_at"),
        Index("ix_cmdb_resource_report_runs_subscription_at", "subscription_id", "created_at"),
        Index("ix_cmdb_resource_report_runs_created_at", "created_at"),
        Index(
            "ix_cmdb_resource_report_runs_person_fact_at",
            "recipient_person_id", "fact_hash", "generated_at",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    subscription_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("cmdb_resource_report_subscriptions.id", ondelete="SET NULL"),
    )
    recipient_person_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="CASCADE"), nullable=False,
    )
    trigger: Mapped[str] = mapped_column(String(16), nullable=False)
    status: Mapped[str] = mapped_column(String(24), nullable=False, default="generated")
    window_start: Mapped[date_type] = mapped_column(Date, nullable=False)
    window_end: Mapped[date_type] = mapped_column(Date, nullable=False)
    scope_snapshot: Mapped[dict | None] = mapped_column(JSON)
    report_data: Mapped[dict] = mapped_column(JSON, nullable=False)
    interpretation: Mapped[str | None] = mapped_column(Text)
    ai_interpretation: Mapped[str | None] = mapped_column(Text)
    ai_status: Mapped[str | None] = mapped_column(String(24))
    fact_hash: Mapped[str] = mapped_column(String(64), nullable=False)
    permission_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False)
    error_message: Mapped[str | None] = mapped_column(Text)
    generated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    delivered_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )


class CmdbDomain(Base):
    __tablename__ = "cmdb_domains"

    # The unique constraint (integration_id, domain_name) lives in
    # production since migration 006 — without it the cloud-sync
    # ``mysql_insert(...).on_duplicate_key_update(...)`` path would
    # accumulate duplicate rows on every run. Pre-218 the model did
    # NOT declare it, so ``Base.metadata.create_all`` (used by the
    # pytest schema bootstrap) created the test DB WITHOUT the
    # constraint, and any test that relied on the upsert semantic
    # silently inserted duplicates instead of updating. Mirrored
    # for every sibling sync-upsert table in this module so the
    # ORM-vs-prod drift can't bite a sibling next.
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "domain_name", name="uq_cmdb_domains",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    domain_name: Mapped[str] = mapped_column(String(253), nullable=False)
    registrar: Mapped[str | None] = mapped_column(String(64))
    expiry_date: Mapped[date_type | None] = mapped_column(Date)
    auto_renew: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    icp_filing_no: Mapped[str | None] = mapped_column(String(64))
    status: Mapped[str] = mapped_column(String(32), nullable=False, default="active")
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL")
    )
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Admin-assigned lifecycle classification (prod / staging / dev / test).
    # NULL until explicitly set; not derivable from registrar APIs.
    environment: Mapped[str | None] = mapped_column(String(16))
    # Gate: when False, this domain is hidden from the employee self-service
    # DNS Zone picker.  Default True so existing domains stay requestable.
    allow_dns_request: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="1")
    # Legacy CMDB fields (migration 173).  ``possessor`` is the legal
    # registrant name (distinct from owner_person_id which is the Helix-side
    # person responsible).  ``grade`` is the registrar admin's importance
    # tier — free-form string, no enum.
    grade: Mapped[str | None] = mapped_column(String(16))
    possessor: Mapped[str | None] = mapped_column(String(255))
    remark: Mapped[str | None] = mapped_column(Text)
    # Modifier tracking — set by the service layer on PATCH / annotate.
    # Distinct from ``updated_at`` which advances on every sync upsert.
    last_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    last_modified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Provenance of the most recent write (migration 218):
    #   ``synced``   — last write came from a cloud driver upsert.
    #   ``migrated`` — legacy GCP CloudSQL → Helix one-shot import via
    #                  ``scripts/migrate_cmdb.py``; expiry / auto_renew /
    #                  registrar are stale (not refreshed from any real
    #                  registrar API). The UI should warn on these rows.
    #   ``manual``   — admin set expiry / auto_renew / registrar through
    #                  the ``DomainAnnotation`` PATCH path. A subsequent
    #                  sync that brings a real cloud value still wins
    #                  (cloud is authoritative) — the flag persists only
    #                  until the next sync upsert.
    # Why VARCHAR(16) not ENUM: TiDB's online schema change on ENUM is
    # materially slower than on VARCHAR; membership is enforced at the
    # Pydantic schema layer (DomainAnnotation / DomainResponse).
    data_source: Mapped[str] = mapped_column(
        String(16), nullable=False, default="synced", server_default="synced",
    )


class CmdbServer(Base):
    __tablename__ = "cmdb_servers"

    # Production has had ``uq_cmdb_servers (integration_id, instance_id)``
    # since migration 006. See CmdbDomain.__table_args__ for the full
    # rationale; recap: ORM model lacked the declaration so the test
    # schema (created via ``Base.metadata.create_all``) was missing the
    # constraint, and sync-upsert tests silently inserted duplicates.
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "instance_id", name="uq_cmdb_servers",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    instance_id: Mapped[str] = mapped_column(String(512), nullable=False)
    name: Mapped[str | None] = mapped_column(String(256))
    status: Mapped[str] = mapped_column(String(32), nullable=False)
    region: Mapped[str | None] = mapped_column(String(64))
    zone: Mapped[str | None] = mapped_column(String(64))
    vcpu: Mapped[int | None] = mapped_column(Integer)
    memory_mb: Mapped[int | None] = mapped_column(Integer)
    public_ips: Mapped[list[str] | None] = mapped_column(JSON)
    private_ips: Mapped[list[str] | None] = mapped_column(JSON)
    os_type: Mapped[str | None] = mapped_column(String(16))
    image_id: Mapped[str | None] = mapped_column(String(128))
    tags: Mapped[dict | None] = mapped_column(JSON)
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL")
    )
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Legacy CMDB / cloud-detail fields (migration 173). ``app`` is the
    # free-text application label; ``remark`` is the admin's notes.
    # ``instance_type``, ``image_name``, ``vpc_id``, ``subnet_id``,
    # ``project_id`` can be filled by future sync drivers or by the
    # legacy import. IDC/bare-metal fields (SN, IPMI, cabinet, rack,
    # switch, model IDs) live in ``raw_data`` JSON — see migration
    # 173 docstring.
    app: Mapped[str | None] = mapped_column(String(255))
    remark: Mapped[str | None] = mapped_column(Text)
    instance_type: Mapped[str | None] = mapped_column(String(64))
    image_name: Mapped[str | None] = mapped_column(String(255))
    vpc_id: Mapped[str | None] = mapped_column(String(128))
    subnet_id: Mapped[str | None] = mapped_column(String(128))
    project_id: Mapped[str | None] = mapped_column(String(128))
    system_disk_gb: Mapped[int | None] = mapped_column(Integer)
    data_disk_gb: Mapped[int | None] = mapped_column(Integer)
    bandwidth_mbps: Mapped[int | None] = mapped_column(Integer)
    expired: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="0")
    # Modifier tracking (migration 173).
    last_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    last_modified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Stamped when the owner responds to a semiannual server-inventory
    # confirmation card (migration 214). The cron skips servers whose
    # ``last_owner_confirmed_at >= now - 180d``; once an owner clicks
    # "确认仍在使用" the row is good for another half year. Null until
    # the first confirmation lands — the cron's first sweep covers
    # all eligible servers anyway, so null behaves identically to a
    # very-old confirmation.
    last_owner_confirmed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )


# ─── Cloud efficiency / cost-optimization (utilization) ─────────────
# Verdicts + recommendations are a closed vocabulary shared by the rule
# engine, router schemas, MCP tools, and the frontend i18n keys. Keep
# them as module constants so a rename can't silently drift between the
# writer (efficiency engine) and the readers.
EFFICIENCY_VERDICT_IDLE = "idle"                 # essentially unused → stop
EFFICIENCY_VERDICT_OVERSIZED = "oversized"       # chronically underused → downsize
EFFICIENCY_VERDICT_HEALTHY = "healthy"           # utilization in a sensible band
EFFICIENCY_VERDICT_BUSY = "busy"                 # hot → keep (maybe undersized)
EFFICIENCY_VERDICT_INSUFFICIENT_DATA = "insufficient_data"  # too few sampled days
EFFICIENCY_VERDICT_NO_METRICS = "no_metrics"     # monitoring returned nothing

EFFICIENCY_RECO_STOP = "stop"
EFFICIENCY_RECO_DOWNSIZE = "downsize"
EFFICIENCY_RECO_KEEP = "keep"
EFFICIENCY_RECO_INVESTIGATE = "investigate"
EFFICIENCY_RECO_NONE = "none"


class CmdbServerMetricsDaily(Base):
    """Per-(server × day) rollup of normalized cloud utilization, written
    by the ``cmdb.server_metrics_sync`` handler from each cloud's
    monitoring API (CloudWatch / CloudMonitor / Azure Monitor / GCP Cloud
    Monitoring). Feeds the efficiency engine's rolling-window verdict and
    the detail-page utilization sparkline. Idempotent upsert on
    ``(server_id, metric_date)``; pruned to ~90 days by the same handler.

    Metric columns are nullable and degrade gracefully. CPU + network land
    on every cloud with no in-guest agent; memory / disk *usage* need a
    guest agent (CloudWatch agent / CloudMonitor agent / Azure diagnostics
    / GCP Ops agent) and stay NULL otherwise. NULL ≠ 0 — a missing
    datapoint must never be read as "0% / idle" by the rule engine."""

    __tablename__ = "cmdb_server_metrics_daily"

    __table_args__ = (
        UniqueConstraint(
            "server_id", "metric_date", name="uq_server_metrics_daily",
        ),
        Index(
            "ix_server_metrics_daily_int_date", "integration_id", "metric_date",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    server_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("cmdb_servers.id", ondelete="CASCADE"),
        nullable=False,
    )
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False,
    )
    metric_date: Mapped[date_type] = mapped_column(Date, nullable=False)
    # Percentages 0–100. Float (not Numeric) so the engine's mean / p95 /
    # max arithmetic stays float-native — money stays Numeric below.
    cpu_pct_avg: Mapped[float | None] = mapped_column(Float)
    cpu_pct_p95: Mapped[float | None] = mapped_column(Float)
    cpu_pct_max: Mapped[float | None] = mapped_column(Float)
    mem_pct_avg: Mapped[float | None] = mapped_column(Float)
    mem_pct_p95: Mapped[float | None] = mapped_column(Float)
    mem_pct_max: Mapped[float | None] = mapped_column(Float)
    disk_used_pct_avg: Mapped[float | None] = mapped_column(Float)
    disk_used_pct_max: Mapped[float | None] = mapped_column(Float)
    net_in_bytes_avg: Mapped[int | None] = mapped_column(BigInteger)
    net_out_bytes_avg: Mapped[int | None] = mapped_column(BigInteger)
    disk_read_bytes_avg: Mapped[int | None] = mapped_column(BigInteger)
    disk_write_bytes_avg: Mapped[int | None] = mapped_column(BigInteger)
    # How many hourly buckets backed this day's rollup — surfaces thin
    # sampling (a stopped instance reports nothing) to the engine.
    sample_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    source: Mapped[str | None] = mapped_column(String(32))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


class CmdbServerEfficiency(Base):
    """One row per server: the rolling-window efficiency profile the
    cost-optimization dashboard, detail-page tab, self-service view, and
    MCP tools all read. Recomputed daily by ``cmdb.efficiency_compute``
    from ``cmdb_server_metrics_daily`` (utilization) + ``cmdb_cost_facts``
    (per-resource spend) + the configurable rule engine.

    Read-only insight: this feature never writes to the cloud. ``verdict``
    and ``recommendation`` are advisory — execution stays in the existing
    ServerLifecycleActions surface. ``ai_narrative`` is filled in a later
    PR by the Bailian interpreter and is orthogonal to the deterministic
    rule verdict (the rules stand on their own without AI)."""

    __tablename__ = "cmdb_server_efficiency"

    __table_args__ = (
        UniqueConstraint("server_id", name="uq_server_efficiency"),
        Index(
            "ix_server_efficiency_int_verdict", "integration_id", "verdict",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    server_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("cmdb_servers.id", ondelete="CASCADE"),
        nullable=False,
    )
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id", ondelete="CASCADE"), nullable=False,
    )
    window_days: Mapped[int] = mapped_column(
        Integer, nullable=False, default=14, server_default="14",
    )
    # Days within the window that actually had samples — drives the
    # ``insufficient_data`` guard so a box monitored for one day never
    # gets a confident "idle → stop".
    data_days: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    cpu_avg: Mapped[float | None] = mapped_column(Float)
    cpu_p95: Mapped[float | None] = mapped_column(Float)
    cpu_max: Mapped[float | None] = mapped_column(Float)
    mem_avg: Mapped[float | None] = mapped_column(Float)
    mem_p95: Mapped[float | None] = mapped_column(Float)
    mem_max: Mapped[float | None] = mapped_column(Float)
    net_in_avg: Mapped[int | None] = mapped_column(BigInteger)
    net_out_avg: Mapped[int | None] = mapped_column(BigInteger)
    verdict: Mapped[str] = mapped_column(
        String(32),
        nullable=False,
        default=EFFICIENCY_VERDICT_INSUFFICIENT_DATA,
        server_default=EFFICIENCY_VERDICT_INSUFFICIENT_DATA,
    )
    # Rule codes that fired (e.g. ["cpu_p95_idle", "net_idle"]) — lets the
    # UI + AI explain *why* without re-deriving.
    verdict_reasons: Mapped[list | None] = mapped_column(JSON)
    recommendation: Mapped[str] = mapped_column(
        String(32),
        nullable=False,
        default=EFFICIENCY_RECO_NONE,
        server_default=EFFICIENCY_RECO_NONE,
    )
    # Free-form structured hint for the recommendation (e.g. suggested
    # smaller class heuristic, target vcpu, savings basis).
    recommended_action_detail: Mapped[dict | None] = mapped_column(JSON)
    # Trailing ~30d spend for this instance, joined from cmdb_cost_facts.
    # NULL when the cost line-item couldn't be matched to this instance —
    # utilization still renders; savings just can't be quantified.
    monthly_cost_usd: Mapped[Decimal | None] = mapped_column(Numeric(14, 2))
    estimated_monthly_savings_usd: Mapped[Decimal | None] = mapped_column(
        Numeric(14, 2),
    )
    cost_currency: Mapped[str | None] = mapped_column(String(8))
    # Spec snapshot at compute time (migration 425). The console-savings
    # detector compares the freshly-synced server spec against this to
    # recognise an out-of-band downsize of a flagged box — sync overwrites
    # cmdb_servers in place, so without the snapshot the old spec is gone.
    instance_type: Mapped[str | None] = mapped_column(String(128))
    vcpu: Mapped[int | None] = mapped_column(Integer)
    memory_mb: Mapped[int | None] = mapped_column(Integer)
    ai_narrative: Mapped[str | None] = mapped_column(Text)
    ai_narrative_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    rule_version: Mapped[str | None] = mapped_column(String(16))
    computed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


# ─── Efficiency recommendation triage (owner participation) ─────────
# Lightweight, NON-DESTRUCTIVE owner response to a per-server
# recommendation. This is the "让大家都参与" layer: the subscription
# owner / server owner records a decision on their own waste so a
# recommendation stops being a read-only dead-end. It NEVER executes a
# cloud action (stop / decommission / downsize stay in the existing
# lifecycle surface); it only records intent + clears the item from the
# owner's "待处理" queue.
EFFICIENCY_TRIAGE_OPEN = "open"              # not yet triaged (implicit default)
EFFICIENCY_TRIAGE_KEPT = "kept"              # owner: keep as-is (reason recorded)
EFFICIENCY_TRIAGE_ACKNOWLEDGED = "acknowledged"  # owner: seen, will handle
EFFICIENCY_TRIAGE_SNOOZED = "snoozed"        # owner: remind me later
EFFICIENCY_TRIAGE_STATUSES = (
    EFFICIENCY_TRIAGE_KEPT,
    EFFICIENCY_TRIAGE_ACKNOWLEDGED,
    EFFICIENCY_TRIAGE_SNOOZED,
)


class CmdbServerEfficiencyTriage(Base):
    """One row per server: the owner's triage decision on that server's
    current efficiency recommendation. Kept in a SEPARATE table from
    ``cmdb_server_efficiency`` so it survives the daily recompute (which
    overwrites the efficiency row) and keeps its own audit timestamps.

    Freshness rule (enforced by the reader / recompute, see
    ``efficiency_service`` + ``efficiency.compute_efficiency_for_integration``):
    a triage is only "live" while the server's current ``verdict`` still
    matches ``verdict_at_decision``. If the daily recompute flips the
    verdict (e.g. idle → busy, or the recommendation changes), the triage
    is treated as expired and the item returns to ``open`` — the owner's
    "keep this idle box" decision must not silently suppress a materially
    different future finding. We compare at read time rather than deleting
    the row, so the decision history + who/when stays auditable."""

    __tablename__ = "cmdb_server_efficiency_triage"

    __table_args__ = (
        UniqueConstraint("server_id", name="uq_server_efficiency_triage"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    server_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("cmdb_servers.id", ondelete="CASCADE"),
        nullable=False,
    )
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default=EFFICIENCY_TRIAGE_ACKNOWLEDGED,
    )
    reason: Mapped[str | None] = mapped_column(Text)
    # The verdict the owner was looking at when they decided. If the
    # server's current verdict drifts from this, the triage is stale.
    verdict_at_decision: Mapped[str | None] = mapped_column(String(32))
    decided_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    decided_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    # Only set when status == snoozed; the item stays suppressed from the
    # "待处理" queue until this instant, then re-surfaces as open.
    snooze_until: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


# ─── Realized cost-optimization savings (outcome ledger) ────────────
# The efficiency engine only ever computed ESTIMATED savings ("if you
# acted, you'd save $X"). This ledger closes the loop: when a server that
# WAS flagged idle/oversized is actually eliminated (destroyed) or stopped,
# a post-log audit hook records the pre-action monthly cost baseline here.
# The REALIZED amount is computed at read time as (baseline − actual
# post-action cost from cmdb_cost_facts) — so a destroy shows the full
# drop, and a machine that was stopped then started back up self-corrects
# to ~0 saved (no overcount, no cron). "Only flagged machines count" is the
# conservative attribution the user chose: an elimination is credited only
# when the server carried an idle/oversized verdict (live or via triage).
SAVINGS_METHOD_DESTROY = "destroy"   # irreversible termination → cost→0
SAVINGS_METHOD_STOP = "stop"         # power_off (reversible; residual storage)
SAVINGS_METHOD_DOWNSIZE = "downsize"  # resized to a smaller spec (console-side)

# Where the qualifying action was observed. ``helix`` = executed inside
# Helix (audit post-log hook on cmdb.server.destroy / power_off);
# ``console`` = done directly in the cloud console and recognised by the
# sync-side detector (savings_detector) — a flagged box that disappeared,
# stopped, or shrank between syncs.
SAVINGS_SOURCE_HELIX = "helix"
SAVINGS_SOURCE_CONSOLE = "console"


class CmdbEfficiencySavingsEvent(Base):
    """One row per server that was flagged and then eliminated/stopped.
    Written by the ``cmdb.savings_tracker`` audit hook; the realized amount
    is derived at read time from ``cmdb_cost_facts`` (see module docstring).
    ``baseline`` is the pre-action trailing monthly cost, frozen at the
    first qualifying action; ``method`` tracks the latest action."""

    __tablename__ = "cmdb_efficiency_savings_events"

    __table_args__ = (
        UniqueConstraint("server_id", name="uq_savings_event_server"),
        Index("ix_savings_event_acted", "acted_at"),
        Index("ix_savings_event_integration", "integration_id"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Plain string (NOT an FK): the server is often soft-deleted /
    # terminated right after the action, and we must keep the outcome even
    # if the row is later hard-pruned.
    server_id: Mapped[str] = mapped_column(String(36), nullable=False)
    instance_id: Mapped[str | None] = mapped_column(String(512))
    integration_id: Mapped[str | None] = mapped_column(String(36))
    # Cost-facts scope attribution snapshot, for the leadership rollup's
    # per-subscription grouping.
    scope_id: Mapped[str | None] = mapped_column(String(36))
    method: Mapped[str] = mapped_column(String(16), nullable=False)
    # 'helix' (audit hook) vs 'console' (sync-side detector) — see the
    # SAVINGS_SOURCE_* vocabulary above. Migration 425.
    source: Mapped[str] = mapped_column(
        String(16), nullable=False, default=SAVINGS_SOURCE_HELIX,
        server_default=SAVINGS_SOURCE_HELIX,
    )
    # Detector breadcrumbs (e.g. from/to instance_type + vcpu/memory for a
    # downsize) so the number stays traceable to "what changed".
    evidence: Mapped[dict | None] = mapped_column(JSON)
    verdict_at_action: Mapped[str | None] = mapped_column(String(32))
    monthly_cost_baseline_usd: Mapped[Decimal | None] = mapped_column(
        Numeric(14, 2),
    )
    cost_currency: Mapped[str | None] = mapped_column(String(8))
    # Owner + department snapshot at action time (for the by-department
    # leadership breakdown; department resolution can drift later).
    owner_person_id: Mapped[str | None] = mapped_column(String(36))
    owner_department: Mapped[str | None] = mapped_column(String(256))
    acted_by_person_id: Mapped[str | None] = mapped_column(String(36))
    acted_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


# Resource types tracked by the unified expiry-alert engine.
EXPIRY_RESOURCE_CERTIFICATE = "certificate"
EXPIRY_RESOURCE_DOMAIN = "domain"


class CmdbExpiryAlertState(Base):
    """Unified per-resource dedup + escalation + acknowledgement state for
    the graduated expiry-alert engine (migration 362), shared by the cert
    and domain expiry crons.

    Replaces the old per-type ``cmdb_cert_expiry_alert_state`` /
    ``cmdb_domain_expiry_alert_state`` tables. See
    ``app.domains.cmdb.expiry_alert_engine`` for the cadence
    (T-30 / T-14 once → daily inside T-7 and while expired), escalation
    (本人 + 域名负责人 + lead in the last-mile window), first-enable backlog
    suppression (via ``escalation_started_on``), and the acknowledge /
    snooze loop (via ``acknowledged_until``)."""

    __tablename__ = "cmdb_expiry_alert_state"

    resource_type: Mapped[str] = mapped_column(String(16), primary_key=True)
    resource_id: Mapped[str] = mapped_column(String(36), primary_key=True)
    last_bucket: Mapped[str] = mapped_column(String(16), nullable=False)
    last_owner_person_id: Mapped[str | None] = mapped_column(String(36))
    # Calendar date of the last ACTUAL send — drives daily dedup inside
    # the escalation window (re-send once per day, not once per bucket).
    last_alerted_on: Mapped[date_type | None] = mapped_column(Date)
    # Non-null ⇒ the resource entered the escalation window (≤7d / expired)
    # under the cron's watch (a go-forward transition) ⇒ eligible for daily
    # escalated sends. NULL ⇒ not in the window, or seeded as backlog at
    # first-enable (suppressed). This is the first-enable storm guard.
    escalation_started_on: Mapped[date_type | None] = mapped_column(Date)
    # Snooze: an owner who marks the resource "renewing" suppresses all
    # alerts (incl. escalation) through this date.
    acknowledged_until: Mapped[date_type | None] = mapped_column(Date)
    acknowledged_by_person_id: Mapped[str | None] = mapped_column(String(36))
    acknowledged_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    last_alerted_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )


# ─── TLS/SSL certificate inventory ──────────────────────────────────
# Provenance of a certificate row. ``probed`` = Helix opened a TLS
# connection and read the served leaf cert; ``cloud`` = synced from a
# cloud cert manager (ACM / Aliyun / Tencent SSL); ``manual`` = admin
# entered it for a cert Helix can't reach over the network.
CERT_SOURCE_PROBED = "probed"
CERT_SOURCE_CLOUD = "cloud"
CERT_SOURCE_MANUAL = "manual"
CERT_SOURCES = frozenset({CERT_SOURCE_PROBED, CERT_SOURCE_CLOUD, CERT_SOURCE_MANUAL})

# Derived lifecycle status. ``valid`` / ``expiring`` / ``expired`` are
# computed from ``not_after`` at probe/sync time; ``unreachable`` means a
# probe target could not be reached (NOT the same as expired — never page
# an owner about an unreachable internal host as if the cert lapsed);
# ``unknown`` is the pre-observation default.
CERT_STATUS_VALID = "valid"
CERT_STATUS_EXPIRING = "expiring"
CERT_STATUS_EXPIRED = "expired"
CERT_STATUS_UNREACHABLE = "unreachable"
CERT_STATUS_UNKNOWN = "unknown"
CERT_STATUSES = frozenset({
    CERT_STATUS_VALID, CERT_STATUS_EXPIRING, CERT_STATUS_EXPIRED,
    CERT_STATUS_UNREACHABLE, CERT_STATUS_UNKNOWN,
})


class CmdbCertificate(Base):
    """One row per observed TLS/SSL certificate (migration 267).

    Idempotent upsert key is ``dedup_key`` (see migration docstring):
    ``probed:{host}:{port}`` / ``cloud:{integration_id}:{cloud_cert_id}``
    / ``manual:{uuid}``. A renewal changes ``fingerprint_sha256`` +
    ``not_after`` but keeps the dedup_key, so the upsert UPDATES the row
    in place and the moved-forward ``not_after`` auto-clears the alert
    state on the next ``cmdb.cert_expiry_alert`` tick.
    """

    __tablename__ = "cmdb_certificates"
    __table_args__ = (
        UniqueConstraint("dedup_key", name="uq_cmdb_certificates_dedup"),
        Index("ix_cmdb_certificates_owner", "owner_person_id"),
        Index("ix_cmdb_certificates_not_after", "not_after"),
        Index("ix_cmdb_certificates_status", "status"),
        Index("ix_cmdb_certificates_integration", "integration_id"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    dedup_key: Mapped[str] = mapped_column(String(512), nullable=False)
    source: Mapped[str] = mapped_column(String(16), nullable=False)
    # NULL for probed / manual; set for cloud-synced certs.
    integration_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("integrations.id", ondelete="CASCADE"),
    )
    cloud_cert_id: Mapped[str | None] = mapped_column(String(256))
    # Probe coordinates (NULL for cloud / manual).
    probe_host: Mapped[str | None] = mapped_column(String(256))
    probe_port: Mapped[int] = mapped_column(
        Integer, nullable=False, default=443, server_default="443",
    )
    # Parsed certificate fields.
    common_name: Mapped[str | None] = mapped_column(String(256))
    sans: Mapped[list | None] = mapped_column(JSON)
    issuer: Mapped[str | None] = mapped_column(String(256))
    serial_number: Mapped[str | None] = mapped_column(String(128))
    fingerprint_sha256: Mapped[str | None] = mapped_column(String(64))
    not_before: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    not_after: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    key_algorithm: Mapped[str | None] = mapped_column(String(32))
    key_bits: Mapped[int | None] = mapped_column(Integer)
    signature_algorithm: Mapped[str | None] = mapped_column(String(64))
    status: Mapped[str] = mapped_column(
        String(24), nullable=False, default=CERT_STATUS_UNKNOWN,
        server_default=CERT_STATUS_UNKNOWN,
    )
    last_error: Mapped[str | None] = mapped_column(Text)
    # NOTE: a certificate does NOT independently own a person. Its
    # responsible owner is LIVE-RESOLVED from the DNS record / zone
    # domain it serves (see ``cert_owner_resolver.CertOwnerResolver``).
    # This column survives only as a deliberate admin OVERRIDE — set it
    # and the resolver honours it; leave ``owner_is_manual=False`` and the
    # resolver ignores any value here and walks the DNS→domain→network
    # chain instead. Legacy auto-snapshot values (pre-migration 360) are
    # left in place but dormant (owner_is_manual=False).
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    # True only when ``owner_person_id`` is a deliberate admin override
    # (manual-entry cert, or explicit re-assignment via PATCH). Migration
    # 360; defaults False so probed/cloud certs always live-resolve.
    owner_is_manual: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="0",
    )
    environment: Mapped[str | None] = mapped_column(String(16))
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    # Set by the cloud-cert sync upsert (source='cloud'); NULL for
    # probed / manual rows. Drives ``_safe_sweep`` mark-and-sweep.
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    last_probed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    remark: Mapped[str | None] = mapped_column(Text)
    last_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    last_modified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class CmdbNetworkOwnerHistory(Base):
    """Append-only history of every owner change on network resources.

    Created by migration 175. Written by the service layer on every
    ``cmdb_domains`` / ``cmdb_dns_records`` / ``cmdb_cdn_domains`` /
    ``cmdb_load_balancers`` owner mutation — PATCH annotate, batch
    reassign, DNS-request execute, offboarding force-transfer, sync
    initial attribution.

    Distinct from ``audit_logs``: that table covers ANY state change;
    this one is a domain-specific append-only log so admin UIs can
    cheaply ask "who has owned this domain over its lifetime?"
    without scanning the full audit table for matching detail JSON.
    """

    __tablename__ = "cmdb_network_owner_history"

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # One of: ``domain`` / ``dns_record`` / ``cdn_domain`` / ``load_balancer``.
    resource_type: Mapped[str] = mapped_column(String(32), nullable=False)
    resource_id: Mapped[str] = mapped_column(String(36), nullable=False)
    previous_owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    new_owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    actor_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    # Why this change happened. Keep aligned with
    # ``app.domains.cmdb.owner_history.CHANGE_REASON_*`` constants.
    change_reason: Mapped[str] = mapped_column(String(64), nullable=False)
    trace_id: Mapped[str | None] = mapped_column(String(40))
    context: Mapped[dict | None] = mapped_column(JSON)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )

    __table_args__ = (
        Index(
            "ix_cmdb_network_owner_history_resource",
            "resource_type", "resource_id", "created_at",
        ),
        Index(
            "ix_cmdb_network_owner_history_prev_owner",
            "previous_owner_person_id",
        ),
        Index(
            "ix_cmdb_network_owner_history_new_owner",
            "new_owner_person_id",
        ),
    )


class CmdbUserMapping(Base):
    __tablename__ = "cmdb_user_mapping"

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    cmdb_username: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
    cmdb_email: Mapped[str | None] = mapped_column(String(255))
    cmdb_is_superuser: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    helix_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL")
    )
    match_status: Mapped[str] = mapped_column(String(32), nullable=False, default="not_in_helix")
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())


class CmdbDnsRecord(Base):
    __tablename__ = "cmdb_dns_records"

    # Production has had ``uq_cmdb_dns_records (integration_id,
    # zone_name, record_id)`` since migration 006. See
    # CmdbDomain.__table_args__ for the full rationale.
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "zone_name", "record_id",
            name="uq_cmdb_dns_records",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    zone_name: Mapped[str] = mapped_column(String(253), nullable=False)
    record_id: Mapped[str] = mapped_column(String(400), nullable=False)
    name: Mapped[str] = mapped_column(String(256), nullable=False)
    record_type: Mapped[str] = mapped_column(String(32), nullable=False)
    value: Mapped[str] = mapped_column(Text, nullable=False)
    ttl: Mapped[int] = mapped_column(Integer, nullable=False, default=600)
    line: Mapped[str | None] = mapped_column(String(64))
    status: Mapped[str] = mapped_column(String(32), nullable=False, default="enable")
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), index=True,
    )
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_owner_review_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Stamped by ``cmdb.dns_owner_recertify`` cron after a successful
    # nag DM lands. Distinct from ``last_owner_review_at`` (set by the
    # owner CONFIRMING). Cron skips records nagged within the dedup
    # window so the owner gets at most one nag per record per period.
    # Added in migration 177.
    last_recertify_nag_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Legacy CMDB field (migration 173).
    remark: Mapped[str | None] = mapped_column(Text)
    # Modifier tracking (migration 173).
    last_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    last_modified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


# Lifecycle states for cmdb_dns_record_requests. Ordered roughly in
# state-machine flow so eyeballing tests + audit logs is cheaper.
DNS_REQ_PENDING = "pending"
DNS_REQ_APPROVED = "approved"
DNS_REQ_REJECTED = "rejected"
DNS_REQ_EXECUTING = "executing"
DNS_REQ_EXECUTED = "executed"
DNS_REQ_FAILED = "failed"
DNS_REQ_CANCELLED = "cancelled"
DNS_REQ_TERMINAL_STATES = frozenset(
    {DNS_REQ_REJECTED, DNS_REQ_EXECUTED, DNS_REQ_FAILED, DNS_REQ_CANCELLED},
)

DNS_REQ_ACTION_CREATE = "create"
DNS_REQ_ACTION_UPDATE = "update"
DNS_REQ_ACTION_DELETE = "delete"
# Toggle the cloud-side record status (DNSPod ``ENABLE``/``DISABLE`` etc).
# Distinct from update because the change-set is just the status flag,
# not any of value/ttl/name — keeping these as separate actions lets the
# Feishu approval card display the right human description and lets the
# executor skip value-format validation that doesn't apply.
DNS_REQ_ACTION_ENABLE = "enable"
DNS_REQ_ACTION_DISABLE = "disable"
DNS_REQ_ACTIONS = frozenset(
    {
        DNS_REQ_ACTION_CREATE,
        DNS_REQ_ACTION_UPDATE,
        DNS_REQ_ACTION_DELETE,
        DNS_REQ_ACTION_ENABLE,
        DNS_REQ_ACTION_DISABLE,
    },
)
# Actions that take a value / TTL / name change as their payload. The
# ``enable`` / ``disable`` actions don't — they reference an existing
# record by ``target_record_id`` and only flip the cloud-side status.
DNS_REQ_VALUE_ACTIONS = frozenset(
    {DNS_REQ_ACTION_CREATE, DNS_REQ_ACTION_UPDATE},
)

# Self-service is intentionally narrower than the admin path. MX / SRV /
# CAA stay admin-only because mis-edits cascade across email delivery (MX),
# service discovery (SRV), or CA issuance authority (CAA) — none of which
# an employee should be able to break with a single approval.
#
# NS is allowed (2026-06-09, see DECISIONS.md): subdomain delegation to an
# external nameserver is a routine self-service need, the value is a plain
# FQDN validated like CNAME, and the same Feishu approval + admin-dispatch
# path gates it. Apex (@) delegation is still effectively guarded by the
# cloud provider, which rejects overwriting a zone's own authoritative NS
# set via the record API.
DNS_REQ_SELF_SERVICE_TYPES = frozenset({"A", "AAAA", "CNAME", "TXT", "NS"})


class CmdbDnsRecordRequest(Base):
    """Employee-submitted DNS create / update / delete request.

    Lives next to :class:`CmdbDnsRecord` instead of inside the asset
    domain because the lifecycle (Feishu approval → Tencent DNSPod
    write) shares zero logic with asset/software requests — the
    overlap is just "rows go through Feishu", which is shallow enough
    that forcing them into the same table would just bloat
    ``AssetEmployeeRequest`` with DNS-only columns.

    Each row is bound to one Feishu instance via
    ``feishu_instance_code`` (UNIQUE, idempotency key for callbacks).
    The state machine is documented on each status constant above.
    """

    __tablename__ = "cmdb_dns_record_requests"
    __table_args__ = (
        UniqueConstraint("feishu_instance_code", name="uq_dns_req_feishu_instance"),
        Index("ix_dns_req_requester", "requester_person_id", "status"),
        Index("ix_dns_req_zone", "zone_name", "status"),
        Index("ix_dns_req_status_created", "status", "created_at"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    requester_person_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("identity_persons.id"), nullable=False,
    )
    action: Mapped[str] = mapped_column(String(16), nullable=False)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False,
    )
    zone_name: Mapped[str] = mapped_column(String(253), nullable=False)
    name: Mapped[str] = mapped_column(String(256), nullable=False)
    record_type: Mapped[str] = mapped_column(String(16), nullable=False)
    value: Mapped[str] = mapped_column(Text, nullable=False)
    ttl: Mapped[int] = mapped_column(Integer, nullable=False, default=600)
    line: Mapped[str | None] = mapped_column(String(64))
    target_record_id: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("cmdb_dns_records.id", ondelete="SET NULL"),
    )
    purpose: Mapped[str] = mapped_column(Text, nullable=False)
    attachment_url: Mapped[str | None] = mapped_column(String(1024))
    status: Mapped[str] = mapped_column(
        String(32), nullable=False, default=DNS_REQ_PENDING,
    )
    feishu_instance_code: Mapped[str | None] = mapped_column(String(128))
    approval_request_id: Mapped[str | None] = mapped_column(String(36))
    executed_record_pk: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("cmdb_dns_records.id", ondelete="SET NULL"),
    )
    error_message: Mapped[str | None] = mapped_column(Text)
    # How many times the executor has entered the cloud-write phase for
    # this row (migration 469). Bounds the executing-reaper's re-arm loop:
    # a row whose execution keeps getting torn down mid-write (deploy
    # storm, provider blackhole) must eventually land in ``failed`` for a
    # human instead of being retried every tick forever.
    execution_attempts: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    submitted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    executed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


class CmdbLoadBalancer(Base):
    __tablename__ = "cmdb_load_balancers"

    # Production has had ``uq_cmdb_load_balancers (integration_id,
    # lb_id)`` since migration 006. See CmdbDomain.__table_args__
    # for the full rationale.
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "lb_id", name="uq_cmdb_load_balancers",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    lb_id: Mapped[str] = mapped_column(String(512), nullable=False)
    name: Mapped[str | None] = mapped_column(String(256))
    lb_type: Mapped[str] = mapped_column(String(32), nullable=False)
    network_type: Mapped[str] = mapped_column(String(32), nullable=False)
    ip_address: Mapped[str | None] = mapped_column(String(256))
    region: Mapped[str | None] = mapped_column(String(64))
    status: Mapped[str] = mapped_column(String(32), nullable=False)
    owner_raw: Mapped[str | None] = mapped_column(String(255))
    # Owner FK (migration 173).  Backfilled from owner_raw on upgrade.
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), index=True,
    )
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Legacy CMDB fields (migration 173).
    app: Mapped[str | None] = mapped_column(String(255))
    remark: Mapped[str | None] = mapped_column(Text)
    tag: Mapped[str | None] = mapped_column(String(255))
    vpc_id: Mapped[str | None] = mapped_column(String(128))
    subnet_id: Mapped[str | None] = mapped_column(String(128))
    project_id: Mapped[str | None] = mapped_column(String(128))
    listener_count: Mapped[int | None] = mapped_column(Integer)
    # Modifier tracking (migration 173).
    last_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    last_modified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class CmdbCdnDomain(Base):
    __tablename__ = "cmdb_cdn_domains"

    # Production has had ``uq_cmdb_cdn_domains (integration_id,
    # domain_name)`` since migration 006. See CmdbDomain.__table_args__
    # for the full rationale.
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "domain_name", name="uq_cmdb_cdn_domains",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    domain_name: Mapped[str] = mapped_column(String(253), nullable=False)
    cname: Mapped[str | None] = mapped_column(String(256))
    status: Mapped[str] = mapped_column(String(32), nullable=False, default="online")
    source_type: Mapped[str | None] = mapped_column(String(32))
    source_address: Mapped[str | None] = mapped_column(String(512))
    is_https: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    owner_raw: Mapped[str | None] = mapped_column(String(255))
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), index=True
    )
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Legacy CMDB fields (migration 173).
    cert_id: Mapped[str | None] = mapped_column(String(128))
    remark: Mapped[str | None] = mapped_column(Text)
    # Modifier tracking (migration 173).
    last_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    last_modified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class CmdbObjectStorage(Base):
    __tablename__ = "cmdb_object_storage"
    __table_args__ = (
        UniqueConstraint("integration_id", "bucket_name", name="uq_oss_account_bucket"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    bucket_name: Mapped[str] = mapped_column(String(256), nullable=False)
    region: Mapped[str | None] = mapped_column(String(64))
    storage_class: Mapped[str | None] = mapped_column(String(64))
    access_type: Mapped[str | None] = mapped_column(String(32))
    size_bytes: Mapped[int | None] = mapped_column(BigInteger)
    object_count: Mapped[int | None] = mapped_column(Integer)
    status: Mapped[str] = mapped_column(String(32), nullable=False, default="active")
    owner_raw: Mapped[str | None] = mapped_column(String(255))
    # Owner FK (migration 366). Brings OSS to parity with LB/CDN/DNS —
    # owner_raw stays as the cloud-tag hint; owner_person_id is the
    # deliberate Helix mapping set by admins via the person picker.
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"), index=True,
    )
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class CmdbIamUser(Base):
    """IAM / sub-account user discovered inside a cloud tenant.

    Populated by the ``iam`` capability sync (AWS IAM ``list_users``,
    future: Aliyun RAM, Tencent CAM). Distinct from ``cmdb_cloud_accounts``
    (which only tracks users IT created for Helix employees) — this
    table shows EVERY IAM user inside the tenant, including service
    accounts, so admins can spot unmanaged accounts.
    """

    __tablename__ = "cmdb_iam_users"
    __table_args__ = (
        # Includes ``sub_account_id`` so providers that iterate
        # sub-accounts (AlicloudIntl Resource Directory, future
        # AWS Organizations cross-account RAM-style listings) don't
        # collapse same-named RAM users from different accounts into
        # a single row. Single-account providers (AWS, Tencent, plain
        # Aliyun) write '' here, behaving exactly as before.
        UniqueConstraint(
            "integration_id", "user_name", "sub_account_id",
            name="uq_cmdb_iam_account_user_subacct",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    user_name: Mapped[str] = mapped_column(String(255), nullable=False)
    # Sub-account / RD-account / org-member id this RAM user belongs to.
    # Empty string for providers without a sub-account concept (AWS IAM,
    # Tencent CAM, single-tenant Aliyun) so the unique key still works
    # without NULL semantics. AlicloudIntl writes the actual member
    # account id (numeric Aliyun AccountId) here.
    sub_account_id: Mapped[str] = mapped_column(
        String(64), nullable=False, default="", server_default="",
    )
    arn: Mapped[str | None] = mapped_column(String(512))
    # AWS/Aliyun return this — informative only.
    user_id_external: Mapped[str | None] = mapped_column(String(512))
    create_date: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Structured attachments so we don't need to parse raw_data at query time.
    attached_policies: Mapped[list | None] = mapped_column(JSON)
    groups: Mapped[list | None] = mapped_column(JSON)
    mfa_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    # What KIND of account this is, once a human has said so:
    # 'human' | 'service' | NULL (unclassified).
    #
    # An IAM user with no ``owner_person_id`` is treated as an orphan on
    # every compliance surface, but two very different things 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 —
    # ``chao.wu@pingcap.cn`` in the first group, ``tiup-server`` and
    # ``operator-v1-cos-upload`` in the second. Without somewhere to
    # record the difference the orphan list can never reach zero, and a
    # list that can never reach zero stops being read.
    account_kind: Mapped[str | None] = mapped_column(String(16))
    account_kind_note: Mapped[str | None] = mapped_column(Text)
    account_kind_set_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    account_kind_set_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # Tri-state companion to ``mfa_enabled``: 'enabled' | 'disabled' |
    # 'unknown' | NULL (provider predates the column).
    #
    # ``mfa_enabled`` is a bool, so a provider that cannot read MFA had
    # only one way to say so — ``False`` — which renders identically to
    # "this user has no MFA". The Tencent provider hard-coded False for
    # all 25 users on a comment claiming CAM exposes per-user MFA only
    # through unstable helpers; it does not (``DescribeSafeAuthFlagColl``
    # takes a ``SubUin``), so the compliance view was reporting a fact it
    # had never asked about. A security surface must be able to say "I
    # don't know" — that is a different remediation from "MFA is off".
    mfa_state: Mapped[str | None] = mapped_column(String(16))
    tags: Mapped[dict | None] = mapped_column(JSON)
    # Mapped Helix person if the IAM user matches one of our employees.
    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    raw_data: Mapped[dict | None] = mapped_column(JSON)
    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    last_synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    # When Helix actively deleted this IAM user on the cloud side
    # (offboarding / scope decommission evacuation). Distinct from
    # ``deleted_at`` (soft-delete) so the IAM users page can render
    # cleaned / revoked rows with a status pill instead of hiding them.
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    revoke_reason: Mapped[str | None] = mapped_column(String(64))
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class CmdbBillingMonthly(Base):
    """One monthly bill per cloud tenant — total + per-service breakdown.

    Written by the ``cmdb.billing_sync`` scheduled job. ``month`` is
    normalized to the first day (UTC) of the billing period; the
    unique constraint lets the job upsert safely when re-running.

    ``cost_by_service`` is a flat dict of ``{service_name: amount}``.
    **Currency caveat (read before aggregating):** ``total_cost_usd`` is
    misnamed — CNY-billed tenants (国内阿里云 / 部分腾讯云 / AliCloud-Intl)
    store the **native CNY** amount here with ``currency='CNY'`` (newer
    rows are rebased to USD at write time with ``currency='USD'``, but
    legacy rows are not). Any reader that aggregates across rows MUST
    convert per-row via ``app.core.fx_rate.CloudCostUsdConverter`` (the
    same correction ``/billing/summary`` and ``/billing/overview`` apply)
    — a bare ``SUM(total_cost_usd)`` mixes USD with ~7×-inflated CNY-as-USD.
    """

    __tablename__ = "cmdb_billing_monthly"
    __table_args__ = (
        UniqueConstraint("integration_id", "month", name="uq_cmdb_billing_account_month"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False
    )
    month: Mapped[date_type] = mapped_column(Date, nullable=False)
    total_cost_usd: Mapped[Decimal] = mapped_column(Numeric(14, 2), nullable=False, default=0)
    cost_by_service: Mapped[dict | None] = mapped_column(JSON)
    currency: Mapped[str] = mapped_column(String(8), nullable=False, default="USD")
    # Raw response kept for auditing / debugging driver parsing mistakes.
    raw_response: Mapped[dict | None] = mapped_column(JSON)
    fetched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


# ─── Cloud account lifecycle ────────────────────────────────────────
# The layered model:
#
#   integrations(kind='cloud')  — tenant credentials (AWS Org master,
#                                 Azure AAD app, GCP org SA, Alicloud RD
#                                 master, or a standalone AWS SG / Tencent
#                                 single-tenant).
#        │
#        ▼  sync'd daily by cmdb.scopes_sync
#   CmdbCloudScope             — sub-account / subscription / project /
#                                 RAM member account / degenerate
#                                 "standalone_iam" row for single-tenant
#                                 clouds. Owner is a Helix person.
#        │
#        ▼  members
#   CmdbCloudAccessGrant       — one row per (scope, person, role) with
#                                 expiry + status. SSO clouds store the
#                                 provider-side principal/assignment id
#                                 in ``external_ref`` / ``detail``; IAM
#                                 clouds store the IAM user name + access
#                                 key pointers in ``detail``.
#
# The older ``cmdb_cloud_accounts`` table (one IAM user per person per
# tenant, per-row offboarding_policy) was dropped in migration 053 —
# it's fully subsumed by the scope/grant model. See DECISIONS.md.


class CmdbCloudScope(Base):
    """Sub-account / subscription / project / RAM-member / standalone.

    One row per billable unit inside a cloud tenant. ``integration_id``
    links back to the tenant credentials row.

    * ``scope_type`` is authoritative for which lifecycle branches apply:
      SSO clouds (aws_account/azure_subscription/gcp_project/aliyun_account)
      expect principals to already exist in the tenant IdP — grants are
      role assignments. ``standalone_iam`` scopes treat the IAM user as
      the unit of grant: one IAM user per person-on-scope, multiple
      policies attached as individual ``CmdbCloudAccessGrant`` rows.

    * ``owner_person_id`` is the "scope owner" — can add/remove members
      without approval (scope-owner RBAC). Unowned scopes produce a
      ``cloud_scope_unowned`` compliance incident AND reject member
      grants until an IT admin assigns an owner.

    * ``budget_usd_monthly``, ``auto_suspend_after_days``,
      ``default_grant_ttl_days`` override the platform-wide defaults
      (``platform_settings`` ``cloud.*``) when set.
    """

    __tablename__ = "cmdb_cloud_scopes"
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "external_id",
            name="uq_cmdb_cloud_scopes_integration_external",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False,
    )
    scope_type: Mapped[str] = mapped_column(String(32), nullable=False)
    external_id: Mapped[str] = mapped_column(String(512), nullable=False)
    display_name: Mapped[str | None] = mapped_column(String(255))

    # Parent scope for hierarchical scope_types (currently only
    # ``azure_resource_group`` → its parent ``azure_subscription``).
    # Tenant-level scopes (the listed members of
    # ``TENANT_LEVEL_SCOPE_TYPES``) always have NULL here. Migration 141.
    parent_scope_id: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("cmdb_cloud_scopes.id", ondelete="SET NULL"),
    )

    owner_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )

    budget_usd_monthly: Mapped[Decimal | None] = mapped_column(Numeric(14, 2))
    # Cost basis ``budget_usd_monthly`` is denominated in (migration 433).
    # ``NULL`` / ``"net"`` = 实付(发票)口径 — the historical default every
    # budget was entered in. ``"list"`` = 业务折前 list 口径, set by
    # scripts/convert_cloud_budgets_to_list_measure when it re-denominates
    # the number by ``list / invoice``.
    #
    # First-class column rather than a platform_settings id list: the basis
    # is a property OF this row, must be readable by the alert evaluator /
    # serializer / UI without a second fetch, and must flip in the SAME
    # transaction as the number it describes. A percentage alert rule's
    # ``threshold_usd`` is ``pct × budget`` by construction, so its
    # ``measure`` stamp is DERIVED from this column — never inherited
    # across a budget edit (that disagreement is #2079's exact shape).
    budget_measure: Mapped[str | None] = mapped_column(String(8))
    auto_suspend_after_days: Mapped[int] = mapped_column(
        Integer, nullable=False, default=60, server_default=text("60"),
    )
    default_grant_ttl_days: Mapped[int] = mapped_column(
        Integer, nullable=False, default=90, server_default=text("90"),
    )

    status: Mapped[str] = mapped_column(
        String(32), nullable=False, default=SCOPE_STATUS_ACTIVE,
        server_default=SCOPE_STATUS_ACTIVE,
    )
    # Lifecycle environment classification (dev / staging / prod). Drives
    # auto-revoke gating: scopes flagged ``prod`` are exempt from the
    # T+0 expiry sweep — production access expiring silently is too
    # disruptive, so the sweeper logs + notifies the owner instead and
    # leaves the grant active until a human renews/revokes explicitly.
    # Dev/staging keep the default revoke-on-expiry behaviour.
    environment: Mapped[str] = mapped_column(
        String(16), nullable=False, default="dev", server_default="dev",
    )
    notes: Mapped[str | None] = mapped_column(Text)
    raw_data: Mapped[dict | None] = mapped_column(JSON)

    # ── Account contact email (migration 450) ───────────────────
    # Cloud providers carry a root/contact email per billable unit
    # (AWS Organizations ``list_accounts`` → ``Email``; other drivers
    # don't expose one yet). ``cloud_account_email`` is the value the
    # driver last reported — CLOUD-AUTHORITATIVE, refreshed on EVERY
    # sync upsert exactly like ``display_name`` (never overwritten by a
    # human). It is read-only in the UI and shown as "云上值".
    #
    # ``account_email_override`` is an owner/admin-set replacement. When
    # set (non-empty) it wins for the "effective email"; clearing it
    # (empty → NULL) reverts to the cloud value. This two-column split
    # is deliberate: the requirement is to display BOTH "what the cloud
    # says" AND "what a human changed it to" side by side, so a single
    # column + a provenance flag (the CmdbServer/CmdbDomain ``data_source``
    # pattern) would lose the cloud value the moment someone overrode it.
    cloud_account_email: Mapped[str | None] = mapped_column(String(320))
    account_email_override: Mapped[str | None] = mapped_column(String(320))
    # Who last set/cleared the override, and when. Stores an
    # ``identity_persons.id`` (resolved to a label in the serializer);
    # plain String — NO FK — mirroring ``decommissioned_by`` on this
    # same table (TiDB-FK-safe; the serializer joins in Python, not SQL,
    # so a hard-deleted person simply renders no label). ``updated_at``
    # is deliberately NOT reused for this — it advances on every sync,
    # which would conflate a human edit with sync activity.
    account_email_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36),
    )
    account_email_modified_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    # NOTE: ``is_ai_dedicated`` was removed in migration 152 (PR #253).
    # AI cost is now identified by service-name classification against
    # the configurable ``ai_spending.service_whitelist`` platform
    # setting — see :mod:`app.domains.ai_spending.whitelist`. The
    # previous tenant-level boolean conflated three real states (100%-
    # AI sub, mixed-workload sub, AI services in a non-AI sub) into
    # one bit and produced over- / under-counting depending on which
    # state the flag mis-described.

    # Per-scope alert rules. Shape:
    #   [{"type": "daily"|"monthly", "threshold_usd": float,
    #     "cc_person_ids": [uuid...], "active": bool,
    #     "last_fired": {"daily:YYYY-MM-DD": true, "monthly:YYYY-MM": true}}]
    # Evaluated by ``cmdb.billing_daily_sync``. See migration 055.
    alert_rules: Mapped[list | None] = mapped_column(JSON)
    # Last date the daily-spike (vs 7-day-avg) alert fired for this
    # scope. Read+set inside ``sync_billing_daily_for_integration``
    # to guarantee at-most-one daily-spike alert per (scope, date) —
    # without this an admin clicking "run sync now" while the cron
    # also fires re-emits the same Feishu card to every owner +
    # IT admin. See migration 114.
    last_spike_alert_date: Mapped[date_type | None] = mapped_column(Date)

    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    last_synced_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )

    # Manual takedown bookkeeping. Set together by ``decommission_scope``
    # alongside ``status='closed'`` only after the cloud-side close API
    # call succeeds (AWS organizations.close_account, Azure
    # subscription cancel, GCP projects.delete, Aliyun RD
    # Delete/RemoveCloudAccount, IAM-mode evacuation). The Helix row
    # never flips to closed unless the provider acknowledged the
    # closure. There is no reactivate path — closures are irreversible.
    #
    # ``decommission_provider_state`` records the post-call cloud-side
    # state so the UI / audit can tell apart ``pending_closure``
    # (90-day grace), ``cancelled``, ``delete_requested``,
    # ``removed_from_org``, ``users_evacuated``, ``not_in_org``, and
    # so on. It's free-form per driver but every value is documented in
    # ``CloseScopeResult.provider_state``.
    decommissioned_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    decommissioned_by: Mapped[str | None] = mapped_column(String(36))
    decommission_reason: Mapped[str | None] = mapped_column(String(500))
    decommission_provider_state: Mapped[str | None] = mapped_column(
        String(64),
    )

    # ── In-flight decommission (migration 624) ────────────────
    #
    # Set when the admin confirmed a decommission that the provider
    # cannot accept YET because the scope still holds blockers Helix is
    # allowed to delete (Azure ``ResourcesExistingOnSubscription``, GCP
    # ``liens_exist``). Helix deletes them and retries the close from
    # ``cmdb.decommission_cleanup_resume`` — the admin's interaction is
    # over at the confirm dialog.
    #
    # ``status`` deliberately stays ``active`` for the whole window.
    # Adding a ``decommissioning`` value to the status enum would have
    # rippled into every predicate that reads it (grant requests, idle
    # sweep, cost attribution, ~7 frontend surfaces), and most of those
    # are asking a different question than "is a takedown in flight".
    # The pending state lives in these columns instead; the queries that
    # genuinely care read ``decommission_pending_since IS NOT NULL``.
    #
    # All six are cleared together when the close finally succeeds (the
    # normal ``decommissioned_*`` columns take over) or when an admin
    # cancels the pending decommission.
    decommission_pending_since: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    decommission_pending_by: Mapped[str | None] = mapped_column(String(36))
    decommission_pending_reason: Mapped[str | None] = mapped_column(
        String(500),
    )
    # The blocker keys the admin was SHOWN and confirmed. The cleanup
    # never touches anything outside this list: a resource group created
    # after the dialog was rendered is not in it, so it cannot be
    # swept up silently — the resume job stops and asks for a fresh
    # confirmation instead. This column is that consent, persisted.
    decommission_cleanup_keys: Mapped[list | None] = mapped_column(JSON)
    # ``cleaning`` — deletes issued, waiting on the provider.
    # ``needs_reconfirm`` — a blocker appeared that nobody confirmed.
    # ``blocked`` — a blocker Helix cannot delete (lock it lacks
    #   permission on, billing attachment); a human was notified once.
    # ``stalled`` — retry budget spent; a human was notified once.
    decommission_cleanup_state: Mapped[str | None] = mapped_column(String(32))
    decommission_cleanup_attempts: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default="0",
    )
    decommission_cleanup_error: Mapped[str | None] = mapped_column(String(500))
    # Trace continuity: the resume job's audit rows carry the trace_id
    # of the click that started it, so one decommission is one trace
    # from confirm dialog to provider acknowledgement days later.
    decommission_cleanup_trace_id: Mapped[str | None] = mapped_column(
        String(64),
    )

    # Source of truth for "owner has confirmed they still manage this
    # scope" — read by the daily ``cmdb.owner_semiannual_confirmation``
    # cron to find scopes whose owners are >180 days past their last
    # confirmation. Updated by the owner-clicks-confirm endpoint AND by
    # ``assign_scope_owner`` on transfer (the new owner implicitly
    # restarts the 180-day clock). NULL means "never confirmed" — cron
    # treats that the same as past-due. Migration 168.
    last_owner_confirmed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    # Per-scope dedup state for ``cmdb.scope_idle_sweep`` (migration 199).
    # Stamped by the weekly sweep whenever a notification is dispatched
    # for this scope; the next sweep skips scopes still inside the
    # ``IDLE_SWEEP_DEDUP_DAYS`` window. NULL = never notified = safe to
    # send. Resets implicitly: once the scope is no longer idle (any
    # grant or spend reappears), it falls out of the candidate query so
    # the column simply stops being read.
    last_idle_notified_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    # Owner-set "I know, ignore until …" timestamp. Set by the
    # ``POST /cmdb/scopes/{id}/snooze-idle`` endpoint (default +30d).
    # Distinct from ``last_idle_notified_at`` — the latter is automatic
    # backoff after a send, this one is an explicit human decision to
    # suppress notifications for longer. Sweep filter is
    # ``COALESCE(idle_snoozed_until, '1970-01-01') > now()``.
    idle_snoozed_until: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    # Backoff marker for the ``cmdb.auto_owner_grant_backfill`` cron when
    # the auto-Owner permission-set is DETERMINISTICALLY unresolvable —
    # i.e. ``_resolve_default_owner_role`` listed the provider's role
    # catalog and found no permission set matching the configured name
    # (distinct from a transient LockBusy / driver crash, which is NOT
    # stamped). Without this marker no ``source='helix'`` ACTIVE grant is
    # ever written, so the cron's ``~sub_helix_owner_grant`` predicate
    # keeps the scope a candidate and re-runs the (paid) Identity Center
    # ``list_scope_roles`` call every 10-min tick forever. Stamped by
    # ``_post_provision_finalize``; the candidate scan skips scopes
    # stamped within ``_OWNER_ROLE_UNRESOLVABLE_RETRY_HOURS`` so an admin
    # who renamed the permission set (in Identity Center or via the
    # ``cloud.aws.scope_request_default_owner_role_name`` platform
    # setting) recovers automatically on the next stale re-scan. Cleared
    # immediately by ``provision_initial_owner_grant`` (a successful
    # resolve+dispatch) and by ``assign_scope_owner`` (manual operator
    # intervention). NULL = resolvable / never attempted. Migration 244.
    auto_owner_grant_unresolvable_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class CmdbCloudAccessGrant(Base):
    """Single user's access to a single scope with a single role.

    A person with multiple roles on one scope gets one row per role
    (uniqueness on ``(scope_id, person_id, role_key)``) — that way
    removing Contributor doesn't implicitly also remove Reader.

    * ``status`` is the state-machine: pending → active → expired /
      revoked / disabled. ``pending`` exists for the Helix-internal
      approval flow (employee submits → scope owner approves → active).
      Owner-direct adds skip ``pending`` and land in ``active`` directly.

    * ``expires_at`` is absolute. Self-service renewal bumps it by
      ``cloud.grant_renewal_days`` (default 90) from NOW — so the clock
      always resets to today + 90 regardless of when they renewed.

    * ``last_notified_stage`` is the idempotency marker for the T-14 /
      T-7 / T-1 reminder schedule: never send the same stage twice per
      (grant, expiry window).

    * ``external_ref`` + ``detail`` carry whatever the driver needs to
      undo: an IAM Identity Center assignment ID, an Azure role
      assignment GUID, a GCP bound IAM member, a RAM user+policy pair,
      or an IAM user name + access key id for standalone_iam scopes.
    """

    __tablename__ = "cmdb_cloud_access_grants"
    __table_args__ = (
        UniqueConstraint(
            "scope_id", "person_id", "role_key",
            name="uq_cmdb_cloud_grants_scope_person_role",
        ),
        # Orphan-cleanup-history idempotency (migration 131). Catches
        # the bug-window where a driver glitch fails to return
        # ``already_absent=True`` on a re-cleanup; without this a
        # second run would insert a duplicate history row because
        # NULL ``person_id`` is distinct under the (scope, person,
        # role) unique key. NULL ``external_principal_id`` makes this
        # constraint a no-op for normal helix-issued / discovered
        # rows.
        UniqueConstraint(
            "scope_id", "role_key", "external_principal_id",
            name="uq_cmdb_cloud_grants_scope_role_extprincipal",
        ),
        Index("ix_cmdb_cloud_grants_expires", "status", "expires_at"),
        Index("ix_cmdb_cloud_grants_person", "person_id", "status"),
        Index(
            "ix_cmdb_cloud_grants_status_reason",
            "status", "revoke_reason",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # NB: ``ondelete=CASCADE`` is a time-bomb that's safe ONLY while
    # the codebase soft-deletes scopes (sets ``status='closed'`` /
    # ``deleted_at=now()``) and never hard-deletes. If a future PR
    # introduces ``session.delete(scope)`` or an ``op.execute("DELETE
    # FROM cmdb_cloud_scopes ...")`` migration, EVERY grant + revoke
    # history row attached to that scope is wiped — silently breaking
    # audit + the 已被撤销 tab + leaver T+1 verify. The pre-launch
    # round-N audit flagged this; review every PR that mutates this
    # table for hard-delete paths before merging. Switching to
    # ``ondelete='SET NULL'`` would protect history but requires
    # making ``scope_id`` nullable + every reader handling None,
    # which is out of scope for the pre-launch hardening pass.
    scope_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("cmdb_cloud_scopes.id", ondelete="CASCADE"),
        nullable=False,
    )
    # Nullable since migration 131 to support cloud-side orphan
    # cleanup history rows (revoke_reason='orphan_principal_swept').
    # When NULL, the principal exists only on the cloud (a stale
    # Identity Center assignment for a deleted SCIM user) — the AWS
    # principal UUID lives in ``external_principal_id`` and a
    # best-effort recovered email in ``external_principal_email``.
    # Every helix-issued and discovered-resolved row still has a
    # populated ``person_id``; only sweep history rows are NULL.
    person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id"), nullable=True,
    )
    role_key: Mapped[str] = mapped_column(String(512), nullable=False)
    role_display_name: Mapped[str | None] = mapped_column(String(255))

    # Set only when ``person_id IS NULL`` (orphan history rows). The
    # cloud-side principal id (AWS Identity Store user UUID, etc.)
    # plus a best-effort recovered email so the admin "已结束" tab
    # can still surface "who" was deleted, even though the Helix
    # person row is gone.
    # ``external_principal_id`` is 128 (not 255) so the orphan
    # uniqueness key ``(scope_id, role_key, external_principal_id)``
    # fits the MySQL/TiDB 3072-byte composite-key cap with role_key
    # index-prefixed to 191 chars. See migration 131.
    external_principal_id: Mapped[str | None] = mapped_column(String(128))
    external_principal_email: Mapped[str | None] = mapped_column(String(255))

    status: Mapped[str] = mapped_column(
        String(32), nullable=False, default=GRANT_STATUS_PENDING,
    )

    granted_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_renewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_notified_stage: Mapped[str | None] = mapped_column(String(16))
    last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    granted_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    employee_request_id: Mapped[str | None] = mapped_column(String(36))

    # Groups multi-role single-submission requests so approve/reject act
    # on the whole group at once. NULL for legacy single-role requests
    # and for cloud-side discovered grants. Indexed for the group-lookup
    # path on approve/reject.
    request_group_id: Mapped[str | None] = mapped_column(String(36))

    external_ref: Mapped[str | None] = mapped_column(String(512))
    detail: Mapped[dict | None] = mapped_column(JSON)

    # Provenance:
    #   ``helix``      — issued via the Helix approval/self-service flow
    #   ``discovered`` — found on the cloud side by ``cmdb.grants_sync``
    #                    and back-filled (no Helix approval trail)
    #   ``imported``   — bulk migrated from legacy systems (rare)
    # Drift detection treats ``helix`` rows that disappear cloud-side
    # more seriously than ``discovered`` rows that disappear.
    source: Mapped[str] = mapped_column(
        String(16), nullable=False, default="helix",
    )
    # Mark-and-sweep for grants_sync: rows with ``source='discovered'``
    # whose ``last_seen_grants_sync_job_id`` doesn't match the current
    # run are candidates for ``drift_removed`` compliance incidents.
    last_seen_grants_sync_job_id: Mapped[str | None] = mapped_column(
        String(36),
    )
    last_seen_grants_sync_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    revoked_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    revoke_reason: Mapped[str | None] = mapped_column(String(64))

    # Role-change review (migration 233, issue #623). When the holder
    # changes department / title / manager, every active grant they
    # hold is flipped to ``review_status='pending_review'`` by
    # ``role_change_coordinator``; a scope owner / IT admin then
    # resolves to ``confirmed_keep`` or ``confirmed_revoke`` from
    # /admin/role-changes. The 30-day sweep cron escalates rows that
    # have rotted at ``pending_review``. ``review_status_changed_at``
    # is the exact moment the status was last written — used by the
    # sweep filter so the 30-day window is precise (NOT ``updated_at``,
    # which bumps on every unrelated grant edit).
    review_status: Mapped[str | None] = mapped_column(String(32))
    review_status_changed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


class CmdbCloudGrantRenewalRequest(Base):
    """A request to extend (or revive) ONE cloud access grant, pending a
    scope-owner decision.

    Why a separate table rather than reusing the grant row's ``pending``
    status: the grant is still ACTIVE while the renewal is under review.
    Flipping it to ``pending`` would revoke the holder's access the moment
    they asked to keep it, and overloading extra columns onto the grant
    row would make "is this grant live?" depend on parsing renewal state.
    A renewal is an event about a grant, so it gets its own row.

    Lifecycle::

        pending_owner ──approve(any ttl ≤ hard cap)──► approved
              ├──reject──► rejected
              ├──cancel──► cancelled   (requester withdrew, or the grant
              │                         was hard-revoked / scope closed)
              └──T+14 cron──► lapsed

    Until 2026-08-26 a TTL above 90 days inserted a ``pending_admin``
    second-signature hop between the owner's approval and ``approved``.
    That tier is gone: the owner's single signature is final at every
    grantable TTL.

    ``approved`` is the only status that moves the grant's ``expires_at``;
    it stamps ``applied_expires_at`` with the value written so the ledger
    can show what the decision actually bought without re-deriving it.

    At most ONE open request may exist per grant, enforced in the database
    rather than by a service-layer check: ``open_for_grant_id`` mirrors
    ``grant_id`` while the request is open and is set to NULL the moment it
    reaches a terminal status. A UNIQUE index over that column therefore
    admits exactly one open row per grant and unlimited closed ones (MySQL
    and TiDB both allow repeated NULLs in a unique index). Without the DB
    guarantee, two clicks racing on the reminder card would create two
    open requests and the owner would see the same person twice in their
    queue.
    """

    __tablename__ = "cmdb_cloud_grant_renewal_requests"
    __table_args__ = (
        UniqueConstraint(
            "open_for_grant_id",
            name="uq_cmdb_grant_renewal_open_per_grant",
        ),
        # Owner queue: "every open renewal on the scopes I own".
        Index(
            "ix_cmdb_grant_renewal_scope_status",
            "scope_id", "status",
        ),
        # Requester's own history for /my/requests.
        Index(
            "ix_cmdb_grant_renewal_person_status",
            "person_id", "status",
        ),
        # Lapse cron: open rows whose deadline has passed.
        Index(
            "ix_cmdb_grant_renewal_lapse",
            "status", "lapse_at",
        ),
        # Group approve/reject acts on every row of one submission.
        Index("ix_cmdb_grant_renewal_group", "request_group_id"),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True)

    # CASCADE mirrors the grant table's own scope CASCADE: renewal history
    # is meaningless once the grant row it describes is gone, and the
    # codebase soft-deletes grants (status flips) rather than hard-deleting
    # them, so in practice this never fires.
    grant_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("cmdb_cloud_access_grants.id", ondelete="CASCADE"),
        nullable=False,
    )
    # Mirrors ``grant_id`` while open, NULL once terminal. See the class
    # docstring — this is the one-open-request-per-grant guarantee.
    open_for_grant_id: Mapped[str | None] = mapped_column(String(36))

    # Denormalized from the grant so the owner queue and the requester's
    # history are single-table index scans. Both are immutable for the
    # life of a grant, so they cannot drift.
    scope_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("cmdb_cloud_scopes.id", ondelete="CASCADE"),
        nullable=False,
    )
    person_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("identity_persons.id"), nullable=False,
    )

    # Who actually submitted. Normally == ``person_id``; differs when a
    # cloud admin raises the request on a holder's behalf (break-glass,
    # e.g. the holder is mid-incident and locked out). Kept distinct so
    # the audit trail never implies the holder asked when they did not.
    requested_by_person_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("identity_persons.id"), nullable=False,
    )

    requested_ttl_days: Mapped[int] = mapped_column(Integer, nullable=False)
    reason: Mapped[str] = mapped_column(String(1000), nullable=False)

    # Set when the request was raised against a grant that had ALREADY
    # been recycled, i.e. approving it re-dispatches access rather than
    # just moving a date. Surfaced prominently to the approver: reviving
    # closed access deserves more scrutiny than extending live access.
    is_revival: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False,
    )

    status: Mapped[str] = mapped_column(
        String(32), nullable=False, default=RENEWAL_STATUS_PENDING_OWNER,
    )
    # Groups one submission covering several of the holder's roles on the
    # same scope, so the owner approves the person once instead of once
    # per role. NULL only for rows written before grouping existed.
    request_group_id: Mapped[str | None] = mapped_column(String(36))

    lapse_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    # Stamped when the approver was nudged that this request is about to
    # lapse. Idempotency marker for that nudge — without it the daily
    # sweep would re-send every morning for the last three days of the
    # window, which is the per-grant reminder-spam pattern this whole
    # rework exists to remove. NULL = not yet nudged.
    owner_reminded_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    # The deciding approver — scope owner, or the cloud-admin team on an
    # ownerless scope (matching how the initial access request routes).
    owner_decided_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    owner_decided_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # Retired second signature (see RENEWAL_STATUS_PENDING_ADMIN). Kept as
    # columns because approved history rows carry values here; nothing
    # written after 2026-08-26. Dropping them needs expand/contract and
    # buys nothing.
    admin_decided_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    admin_decided_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    decision_note: Mapped[str | None] = mapped_column(String(1000))

    # The ``expires_at`` actually written onto the grant on approval.
    # NULL for a perpetual holder (cloud admin) or for any non-approved
    # terminal status.
    applied_expires_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )


class CmdbCloudRole(Base):
    """Provider-side role catalog cached per tenant ± scope.

    Populated daily by ``cmdb.roles_sync``. The request form reads from
    this table (not a live API call) so the dropdown renders fast; the
    "refresh roles" button force-runs the sync on demand.

    ``scope_id`` is nullable because some providers define roles at the
    tenant level (AWS Identity Center permission sets are tenant-wide
    but get *bound* to accounts at assignment time). Others (GCP
    custom roles, Alicloud RAM policies) are scope-local. The sync
    handler writes whichever granularity the provider uses; the request
    form looks up rows matching (integration_id, scope_id) OR
    (integration_id, scope_id IS NULL) and unions them.
    """

    __tablename__ = "cmdb_cloud_roles"
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "scope_id", "role_key",
            name="uq_cmdb_cloud_roles_integration_scope_role",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False,
    )
    scope_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("cmdb_cloud_scopes.id", ondelete="CASCADE"),
    )
    role_key: Mapped[str] = mapped_column(String(512), nullable=False)
    display_name: Mapped[str | None] = mapped_column(String(255))
    description: Mapped[str | None] = mapped_column(Text)
    # Admin-edited description. When set it overrides the provider-synced
    # ``description`` everywhere (catalog UI + employee dropdown); role
    # sync keeps refreshing ``description`` but never touches this. Mirror
    # of the ``is_promoted`` / ``governance_overridden_by_admin`` rule.
    description_override: Mapped[str | None] = mapped_column(Text)
    is_custom: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    source: Mapped[str | None] = mapped_column(String(32))
    raw_data: Mapped[dict | None] = mapped_column(JSON)

    # Whether this role appears in the request modal's default dropdown.
    # Driver classification: TRUE for the admin-curated subset and every
    # tenant-defined custom role; FALSE for vendor built-ins outside the
    # curated subset (Azure has ~380 built-ins, Aliyun has ~250 system
    # policies — surfacing all of them in the employee request modal
    # creates choice paralysis). Admin-flippable via the role catalog
    # promote/demote endpoints.
    is_promoted: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True,
    )

    # TRUE once an admin promotes/demotes this role in the catalog UI.
    # ``is_promoted`` alone was NOT enough to make that decision durable:
    # it lives per (integration, scope, role_key) row, so a demote only
    # touched the rows that existed at that moment, and every scope
    # discovered LATER got a fresh row seeded from the driver default
    # (``is_custom`` ⇒ promoted). Prod 2026-06-11: ten DBaaS-Prod-*
    # CloudSSO roles were demoted across 21 Aliyun sub-accounts and came
    # back promoted on the 8 sub-accounts onboarded afterwards — while
    # the catalog table kept showing 否 (it reports the consensus across
    # scopes), so the admin had no way to see it. This flag is the
    # admin's decision at the (integration, role_key) level: role sync
    # seeds NEW rows from it and re-stamps rows that drifted, so the
    # decision holds for sub-accounts that do not exist yet. Mirrors
    # ``governance_overridden_by_admin``; cleared by 「重置为默认」.
    promotion_overridden_by_admin: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="0",
    )

    # Manual display rank for the catalog table + employee request
    # dropdown. Lower sorts earlier; ``0`` means "unranked" and the
    # existing promoted/custom/alphabetical tiebreakers apply. Admin-set,
    # never recomputed by sync.
    sort_order: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )

    last_seen_sync_job_id: Mapped[str | None] = mapped_column(String(36))
    first_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    last_synced_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # Governance tier — drives visibility / request / approval gating
    # (see app/integrations/cloud_role_governance.py). Computed by the
    # driver classifier during role sync; ``normal`` for low-risk roles.
    governance_tier: Mapped[str] = mapped_column(
        String(16), nullable=False, default="normal", server_default="normal",
    )
    # Human-readable classification reason (which rule fired) — for the
    # admin catalog UI + audit.
    governance_reason: Mapped[str | None] = mapped_column(String(255))
    # TRUE once an admin manually overrides the tier in the catalog UI —
    # role sync then stops recomputing it (mirrors ``is_promoted``).
    governance_overridden_by_admin: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default="0",
    )


class CmdbCloudBillingDaily(Base):
    """Daily billing row keyed by (integration, scope, date).

    Supplements the existing ``CmdbBillingMonthly`` (coarse) with
    per-day, per-scope granularity used for:
      * the scope-detail drill-down chart,
      * the 日环比异常 alert (``daily_spike_multiplier`` + ``_min_usd``
        platform settings),
      * per-member view on ``/my/cloud/scopes/:id`` (members only see
        the scope they're on — cloud billing is not per-user, it's
        per-scope).

    ``scope_id`` nullable to carry the tenant-level aggregate on
    clouds that only expose the tenant total (none of our current
    providers do — included for future-proofing).
    """

    __tablename__ = "cmdb_cloud_billing_daily"
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "scope_id", "date",
            name="uq_cmdb_cloud_billing_daily_key",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False,
    )
    scope_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("cmdb_cloud_scopes.id", ondelete="CASCADE"),
    )
    # Cloud provider's account identifier for this row.
    #   * Registered-scope rows: equals scope.external_id (e.g. 12-digit AWS account).
    #   * Unregistered-account rows (scope_id NULL): the raw account ID.
    #   * Tenant-aggregate / non-per-account providers: '' (empty string).
    # NOT NULL DEFAULT '' so ON DUPLICATE KEY UPDATE works correctly even when
    # scope_id is NULL (MySQL/TiDB treats NULL != NULL in unique indexes).
    linked_account_id: Mapped[str] = mapped_column(
        String(36), nullable=False, server_default="",
    )
    date: Mapped[date_type] = mapped_column(Date, nullable=False)
    total_cost_usd: Mapped[Decimal] = mapped_column(
        Numeric(14, 4), nullable=False, default=0,
    )
    # 业务折前成本（研发口径，list cost）：白名单行的实际执行价合计——
    # spot/阶梯价保留，公司集中折扣（SP/RI/CUD/EDP/赠金/调账/税）剥离。
    # NULL = 该 driver/日期尚未产出双口径数据（迁移 429 之前的历史行，
    # 或该云的 list 规则尚未接入）。读侧研发面 SUM 此列即可，无需行过滤。
    total_cost_usd_list: Mapped[Decimal | None] = mapped_column(Numeric(14, 4))
    # 按需公开牌价合计（AWS=pricing_public；GCP 与 list 同值）。财务页
    # 「研发自主节省」= public − list。NULL 语义同上。
    total_cost_usd_public: Mapped[Decimal | None] = mapped_column(Numeric(14, 4))
    cost_by_service: Mapped[dict | None] = mapped_column(JSON)
    currency: Mapped[str] = mapped_column(String(8), nullable=False, default="USD")
    raw_response: Mapped[dict | None] = mapped_column(JSON)
    # iamPrincipal/Owner (AWS) or equivalent breakdown: {owner_tag_value: cost_usd}.
    # NULL = driver did not perform a cost-allocation-tag query for this row.
    owner_breakdown: Mapped[dict | None] = mapped_column(JSON)

    fetched_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
    )


# ── Scope-creation request (employee → Feishu approval → cloud API) ──

SCOPE_REQUEST_STATUS_PENDING = "pending"
SCOPE_REQUEST_STATUS_FEISHU_SENT = "feishu_sent"
SCOPE_REQUEST_STATUS_FEISHU_APPROVED = "feishu_approved"
SCOPE_REQUEST_STATUS_PROVISIONING = "provisioning"
SCOPE_REQUEST_STATUS_PROVISIONED = "provisioned"
SCOPE_REQUEST_STATUS_REJECTED = "rejected"
SCOPE_REQUEST_STATUS_WITHDRAWN = "withdrawn"
SCOPE_REQUEST_STATUS_FAILED = "failed"


class CmdbCloudScopeRequest(Base):
    """Self-service "create me a new sub-account / subscription" record.

    An employee submits ``provider/integration_id`` + ``name`` +
    ``description`` + ``monthly_budget_usd``. The request flows through
    Feishu approval (the schema_code is configured per-Feishu-integration
    via the ``cloud_scope_request_schema_code`` field). On approval the
    worker calls the cloud driver's ``create_scope`` and inserts the
    resulting :class:`CmdbCloudScope` row with the requester as owner,
    pre-applies the platform default monthly budget + alert rules, and
    grants the requester an "owner-equivalent" role.

    Status state machine:
    ::

        pending → feishu_sent → feishu_approved → provisioning →
        provisioned   (terminal; ``created_scope_id`` populated)

        pending → rejected / withdrawn  (terminal)

        provisioning → failed   (terminal-but-retryable; ``failure_reason``
                                 carries the driver error verbatim)

    The state machine never moves backwards. Retrying a ``failed`` row
    creates a new request — keeps the audit trail honest.
    """

    __tablename__ = "cmdb_cloud_scope_requests"
    __table_args__ = (
        Index(
            "ix_cmdb_scope_requests_requester",
            "requester_person_id", "status",
        ),
        Index(
            "ix_cmdb_scope_requests_integration_status",
            "integration_id", "status",
        ),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True)
    requester_person_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("identity_persons.id"), nullable=False,
    )
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False,
    )
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    description: Mapped[str | None] = mapped_column(Text)
    monthly_budget_usd: Mapped[Decimal] = mapped_column(
        Numeric(14, 2), nullable=False,
    )

    status: Mapped[str] = mapped_column(
        String(32), nullable=False, default=SCOPE_REQUEST_STATUS_PENDING,
    )
    feishu_instance_code: Mapped[str | None] = mapped_column(String(64))
    approver_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    approved_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    rejected_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    rejection_reason: Mapped[str | None] = mapped_column(String(500))

    created_scope_id: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("cmdb_cloud_scopes.id", ondelete="SET NULL"),
    )
    provisioned_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    failure_reason: Mapped[str | None] = mapped_column(String(1000))
    metadata_json: Mapped[dict | None] = mapped_column(JSON)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )



# ─── AI resource ledger (migration 078) ─────────────────────────────
# A "ledger row per AI-capable cloud resource" sitting between the
# scope and the per-day cost table. We need this in CMDB (not in the
# ai_spending domain) because the rows are populated by extending the
# Azure cloud driver's CMDB sync surface — and per the existing
# domain rule, anything that lives off a cloud driver's mirror data
# is CMDB's data layer. The ai_spending domain READS from these rows
# but does not own them.


# NOTE: ``CmdbAiResource`` and ``CmdbAiResourceCostDaily`` were removed
# in PR #253 (migrations 150 + 151). The AI Spending dashboard now
# reads AI cost directly from ``CmdbCostFact`` filtered by the
# configurable ``ai_spending.service_whitelist`` platform setting +
# bucket attribution flows through the AK's ``cloud_scope_id`` →
# ``resource_group`` linkage on facts. No second mirror table required.


# ─── Cost facts + allocation (migration 139) ────────────────────────
# Production-grade FinOps surface. Coarse-grained billing (monthly +
# daily-by-scope) lives in ``CmdbBillingMonthly`` and
# ``CmdbCloudBillingDaily``; this section adds the line-item ground
# truth: one fact row per (date × resource × usage_type) with full
# multi-dimensional breakdown (resource_group, resource_id,
# resource_type, region, service, usage_type, billing-side tags),
# plus denormalized cost-allocation columns (person, department,
# cost_center) populated at ingest time from priority-ordered
# ``CmdbCostAllocationRule`` rows.
#
# The ``cmdb.cost_facts_sync`` cron drives ingestion via each cloud
# driver's ``list_cost_facts(start, end, scope=...)`` method. Daily
# pulls run for yesterday; backfill for an arbitrary range goes
# through the same handler with ``config={"start_date":...,
# "end_date":...}``.


class CmdbCostFact(Base):
    """One billing line item per (date × resource × usage_type).

    The ground-truth fact table the cost-analytics dashboard, MCP
    tools, and chargeback reports all read from. Coarser tables
    (``CmdbBillingMonthly``, ``CmdbCloudBillingDaily``) stay as fast
    KPI caches but are derivable from this table.

    Idempotent re-ingestion: ``fact_key`` is sha256 hex of the
    natural key (date|integration|scope|resource_id|usage_type) and
    carries a UNIQUE constraint, so the cron / backfill paths use
    MySQL ``INSERT ... ON DUPLICATE KEY UPDATE`` to refresh amounts
    without duplicating rows when re-run for the same window.

    ``cost_owner_*`` and ``cost_center`` are denormalized — populated
    at ingest by walking the ``CmdbCostAllocationRule`` ladder first
    (admin overrides), then falling back to the fact's
    ``cmdb_cloud_scopes.owner_person_id`` when no rule matched. The
    chosen path is recorded in ``auto_owner_source`` so the dashboard
    can distinguish rule-driven attribution from scope-owner default.
    Recomputable in-place via the admin "重算分摊" button after a rule
    change OR after a scope-ownership change.
    """

    __tablename__ = "cmdb_cost_facts"
    __table_args__ = (
        UniqueConstraint("fact_key", name="uq_cmdb_cost_facts_fact_key"),
        Index("ix_facts_date_int", "date", "integration_id"),
        # The allocation re-stamp streams one accounting day at a time,
        # continuing each day by primary-key cursor. Keep the entire
        # predicate + cursor inside one narrow index range; without this
        # TiDB can scan a multi-million-row historical id gap just to fill
        # one allocation batch (see migration 458).
        Index("ix_facts_date_id", "date", "id"),
        # 覆盖索引(迁移 431,成本口径改造 PR7):财务瀑布/口径对数区
        # 的月度聚合(date 范围 × integration × bucket × 三金额列)全部
        # 走 IndexRangeScan,不再对 840 万行宽行做 TableFullScan——
        # 后者曾打爆 TiDB serverless 单 SQL 内存配额(错误 8175)。
        Index(
            "ix_facts_date_measure",
            "date", "integration_id", "cost_bucket",
            "amount_usd", "amount_usd_list", "amount_usd_public",
        ),
        Index("ix_facts_resource", "resource_id", "date"),
        Index("ix_facts_int_rg_date", "integration_id", "resource_group", "date"),
        Index("ix_facts_int_service_date", "integration_id", "service", "date"),
        Index("ix_facts_owner_date", "cost_owner_person_id", "date"),
        Index("ix_facts_cc_date", "cost_center", "date"),
        # Pins the billing-tab department filter
        # (``cost_owner_department IN (...) AND date BETWEEN ...``) to an
        # index-range scan instead of dragging the full window's rows
        # through a secondary-index-then-filter. Mirrors ix_facts_cc_date
        # for the sibling cost_center allocation column. Migration 292.
        Index("ix_facts_dept_date", "cost_owner_department", "date"),
        # "Which resources does this subscription attribute" — backs the
        # visibility resolve-time materialization + the scope 成本优化 pane's
        # correlated EXISTS. Migration 427 (prod 8176 OOM fix).
        Index("ix_facts_scope_resource", "scope_id", "resource_id"),
        # Powers the "未分摊 KPI" tooltip and the recompute summary's
        # per-source counts. Migration 145.
        Index("ix_facts_source_date", "auto_owner_source", "date"),
        # Pins per-scope per-day reads + the cron's pre-window DELETE
        # to an index-range scan, so a 45k-row/day AWS payer scope no
        # longer drags TiDB across the full day's row payload to apply
        # a secondary-index-then-filter. Migration 181.
        Index("ix_facts_scope_date", "scope_id", "date"),
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    fact_key: Mapped[str] = mapped_column(String(64), nullable=False)
    date: Mapped[date_type] = mapped_column(Date, nullable=False)
    integration_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("integrations.id", ondelete="RESTRICT"),
        nullable=False,
    )
    scope_id: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("cmdb_cloud_scopes.id", ondelete="SET NULL"),
    )
    resource_group: Mapped[str | None] = mapped_column(String(256))
    resource_id: Mapped[str | None] = mapped_column(String(512))
    resource_name: Mapped[str | None] = mapped_column(String(256))
    resource_type: Mapped[str | None] = mapped_column(String(128))
    region: Mapped[str | None] = mapped_column(String(64))
    service: Mapped[str] = mapped_column(
        String(128), nullable=False, default="", server_default="",
    )
    usage_type: Mapped[str | None] = mapped_column(String(256))
    tags: Mapped[dict | None] = mapped_column(JSON)
    amount_usd: Mapped[Decimal] = mapped_column(
        Numeric(14, 6), nullable=False,
        default=Decimal("0"), server_default=text("0"),
    )
    amount_native: Mapped[Decimal] = mapped_column(
        Numeric(14, 6), nullable=False,
        default=Decimal("0"), server_default=text("0"),
    )
    # 折扣前(原价 / list price)成本,与 amount_usd / amount_native 成对。
    # 可空。当前**只有 AWS** 写入(CUR ``pricing_public_on_demand_cost``);
    # GCP(billing-export ``cost``,credits 之前)与 Azure(零售价需 Price
    # Sheet API)的驱动暂不写 gross → 恒 NULL,各拆 follow-up task。
    # AI 大盘按覆盖率聚合,NULL 不伪造折扣。
    amount_usd_gross: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    amount_native_gross: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    # ── 双口径（2026-07 成本口径改造，迁移 429）────────────────────
    # 业务折前成本（研发口径，list cost）。仅白名单 usage 行有值：
    #   AWS  = line_item_type ∈ {Usage, SavingsPlanCoveredUsage} 的
    #          unblended（SP 覆盖行 unblended 即按需等价，已实测）；
    #          RI 覆盖行（DiscountedUsage）用 pricing_public 还原。
    #   GCP  = cost_at_list（cost 列含合同价 −21% 已禁用），白名单
    #          cost_type='regular' AND service ∉ {Support, Invoice}。
    # 其余行（折扣/费用/赠金/调账/税）恒 NULL——研发口径读侧就是
    # SUM(amount_usd_list)，无需行过滤，防止未知行类型泄入。
    amount_usd_list: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    amount_native_list: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    # 按需公开牌价（AWS=pricing_public_on_demand_cost；GCP 与 list 同值）。
    # 财务页「研发自主节省」（spot/阶梯价）= public − list。
    amount_usd_public: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    # 行分类，财务瀑布按此 GROUP BY：usage / commitment_fee /
    # commitment_discount / contract_discount / credit / adjustment /
    # tax / support / other。默认 'usage'（迁移前存量行的正确值；
    # 折扣类存量行由回填重分类）。
    cost_bucket: Mapped[str] = mapped_column(
        String(32), nullable=False, default="usage", server_default="usage",
    )
    currency: Mapped[str] = mapped_column(
        String(8), nullable=False, default="USD", server_default="USD",
    )
    # ``Numeric(30, 6)`` not ``Numeric(20, 6)``: GCP storage SKUs report
    # cumulative byte-seconds at the daily-aggregate level — a single
    # ``Storage Machine Image`` row for one ``us-central1`` project on
    # 2026-05-09 came in at ``1,570,859,566,034,408 byte-seconds``
    # (~1.57e15, equivalent to ~18 GB stored 1 day). The 14-integer-
    # digit ceiling on ``Numeric(20, 6)`` overflowed and the entire
    # ON DUPLICATE KEY UPDATE batch raised ``Out of range value for
    # column 'usage_quantity'``, killing every per-scope GCP cost-facts
    # ingest. Widened to 24-integer-digit headroom (max ~10^24 — covers
    # multi-EiB tenants for years of storage). Money columns
    # (``amount_usd`` / ``amount_native``) stay at Numeric(14, 6) — they
    # don't need this width and a wider money column would change the
    # finance-aggregate display layer.
    usage_quantity: Mapped[Decimal | None] = mapped_column(Numeric(30, 6))
    usage_unit: Mapped[str | None] = mapped_column(String(64))

    cost_owner_person_id: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    cost_owner_department: Mapped[str | None] = mapped_column(String(256))
    cost_center: Mapped[str | None] = mapped_column(String(64))
    allocation_rule_id: Mapped[str | None] = mapped_column(String(36))

    # How ``cost_owner_person_id`` was decided for this row. Stamped by
    # the ingest / recompute path (migration 145):
    #   * ``rule``         — matched a CmdbCostAllocationRule (admin-
    #                        authored override; ``allocation_rule_id``
    #                        is also set)
    #   * ``scope_owner``  — fell back to ``cmdb_cloud_scopes.owner_
    #                        person_id`` because no rule matched but
    #                        the fact had a scope_id with a known
    #                        owner. Covers the default subscription /
    #                        project / sub-account ownership chain.
    #   * ``none``         — no rule matched, AND either ``scope_id``
    #                        is NULL or the scope lacked an owner.
    #                        Surfaces as "未分摊" in the dashboard.
    # Stored as VARCHAR (not ENUM) so adding ``iam_owner`` /
    # ``rg_owner`` later doesn't require an ALTER TYPE.
    auto_owner_source: Mapped[str] = mapped_column(
        String(16), nullable=False, default="none", server_default="none",
    )

    # ``none_as_null=True`` is load-bearing, not tidiness. SQLAlchemy's
    # JSON type defaults to rendering Python ``None`` as the JSON
    # literal ``null`` — a real value — rather than SQL NULL. On
    # 2026-09-02 that turned ``cmdb.cost_facts_raw_purge`` into an
    # infinite loop in production: the sweep issued
    # ``SET raw='null' WHERE id IN (…)``, its own
    # ``WHERE raw IS NOT NULL`` still matched every row it had just
    # "purged", and it re-selected the same 2000 ids until the
    # scheduler killed it at 1800s having freed nothing. Nobody wants a
    # JSON ``null`` in this column — absent raw means absent.
    raw: Mapped[dict | None] = mapped_column(JSON(none_as_null=True))
    fetched_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )


class CmdbBedrockProfileFact(Base):
    """One AWS Bedrock spend row per (date × owner × profile × service).

    The "profile" is the ``UsedBy`` IAM-principal cost-allocation tag —
    a secondary axis that subdivides ONE owner's Bedrock spend into the
    workloads behind it (``investibook-dev`` / ``investibook-regression``
    / ``aiops`` / …). AWS Cost Explorer's GroupBy is capped at 2
    dimensions and ``cmdb_cost_facts`` already spends them on
    (SERVICE × iamPrincipal/Owner), so this finer breakdown can't live
    in that table without either losing a dimension or double-counting
    its totals. It is stored here, ISOLATED, so no aggregation over
    ``cmdb_cost_facts`` (dashboard, billing drawer, per-AK total,
    by-model) can be perturbed.

    Populated by the same cost-facts cron window as ``cmdb_cost_facts``
    (see ``sync_bedrock_profile_facts_for_integration``) so the per-AK
    detail page's by-profile sum reconciles to its ``mtd_usd``. Only
    AWS integrations whose ``cost_facts.aws_profile_tag.<slug>`` setting
    is configured emit these rows.

    Since migration 435 the setting may list SEVERAL tag keys; each row
    records which axis it came from in ``profile_tag_key`` (see the
    column comment). The per-API-Key axis
    (``iamPrincipal/helix-request-number``) is what lets a key pinned to
    $0 by the shared-Owner rule (#1776) show its own spend.

    Idempotent: ``fact_key`` is the sha256 of the natural key
    (date|integration|scope|owner|tag_key|profile|sub_profile|service)
    and carries a
    UNIQUE constraint; the ingest does a per-(scope, date) pre-window
    DELETE then ``INSERT ... ON DUPLICATE KEY UPDATE`` so re-runs
    converge — including across a change to the configured axis list.

    ``profile`` is stored raw (as CE returns it) — case normalisation
    (``aiops`` vs ``AIOPS``) happens at read time so the ingest stays a
    clean 1:1 with CE groups and never overwrites one casing's amount
    with another's.
    """

    __tablename__ = "cmdb_bedrock_profile_facts"
    __table_args__ = (
        UniqueConstraint(
            "fact_key", name="uq_bedrock_profile_facts_fact_key",
        ),
        Index(
            "ix_bpf_int_owner_date",
            "integration_id", "owner_email", "date",
        ),
        Index(
            "ix_bpf_int_scope_date",
            "integration_id", "scope_external_id", "date",
        ),
        # Per-Key read path (migration 435): pick one tag axis' rows for
        # an integration over a month window.
        Index(
            "ix_bpf_int_key_date",
            "integration_id", "profile_tag_key", "date",
        ),
    )

    id: Mapped[int] = mapped_column(
        BigInteger, primary_key=True, autoincrement=True,
    )
    fact_key: Mapped[str] = mapped_column(String(64), nullable=False)
    date: Mapped[date_type] = mapped_column(Date, nullable=False)
    # Plain String(36), no FK — dodges TiDB CHAR/VARCHAR + collation FK
    # strictness (same call finance_spend_facts made).
    integration_id: Mapped[str] = mapped_column(String(36), nullable=False)
    scope_external_id: Mapped[str | None] = mapped_column(String(64))
    owner_email: Mapped[str] = mapped_column(String(256), nullable=False)
    # Which cost-allocation tag axis this row breaks the owner down by
    # (migration 435). ``UsedBy`` = the workload axis this table shipped
    # with; ``iamPrincipal/helix-request-number`` = the per-API-Key axis
    # that gives a sibling of a shared-Owner collision (#1776) its own
    # number instead of a pinned ``$0``. Rows from different axes are
    # each a FULL breakdown of the same owner total — never sum across
    # ``profile_tag_key`` values.
    profile_tag_key: Mapped[str] = mapped_column(
        String(128), nullable=False,
        default="UsedBy", server_default="UsedBy",
    )
    profile: Mapped[str] = mapped_column(String(256), nullable=False)
    # Second breakdown dimension on the per-API-Key axis (migration 452):
    # for a ``profile_tag_key = iamPrincipal/helix-request-number`` row
    # (``profile`` = the AK number) this holds the ``UsedBy`` workload the
    # spend went to, so a per-key row is (date × key × UsedBy × service)
    # and each key gets its OWN workload table instead of the shared
    # person-level one. NULL on the ``UsedBy`` workload axis (where
    # ``profile`` already IS the UsedBy value) and on any pre-452 row.
    # Joins ``fact_key`` (see ``_profile_fact_key``); NULL normalises to
    # the empty component so two-axis rows can't collide.
    sub_profile: Mapped[str | None] = mapped_column(String(256))
    service: Mapped[str] = mapped_column(String(128), nullable=False)
    amount_usd: Mapped[Decimal] = mapped_column(
        Numeric(18, 6), nullable=False,
    )
    amount_native: Mapped[Decimal | None] = mapped_column(Numeric(18, 6))
    currency: Mapped[str] = mapped_column(
        String(8), nullable=False, default="USD", server_default="USD",
    )
    usage_quantity: Mapped[Decimal | None] = mapped_column(Numeric(30, 6))
    usage_unit: Mapped[str | None] = mapped_column(String(64))
    fetched_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )


class CmdbCostAllocationRule(Base):
    """Priority-ordered rule for attributing cost facts to an owner.

    The ingestor walks rules in ascending ``priority`` (lower number =
    higher priority) where ``enabled=True``, and stamps the first
    matching rule's assignment columns onto the fact row. Rules with
    all-NULL match conditions match every fact (use carefully — they
    typically have the highest ``priority`` value as a catch-all).

    A rule MUST have at least one of ``assign_person_id`` /
    ``assign_department`` / ``assign_cost_center`` set (the writer
    enforces this) — otherwise the row is silently skipped, which
    would surprise admins.
    """

    __tablename__ = "cmdb_cost_allocation_rules"
    __table_args__ = (
        Index("ix_alloc_rules_priority", "enabled", "priority"),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    priority: Mapped[int] = mapped_column(
        Integer, nullable=False, default=100, server_default=text("100"),
    )

    match_integration_ids: Mapped[list | None] = mapped_column(JSON)
    match_provider: Mapped[str | None] = mapped_column(String(32))
    match_resource_group: Mapped[str | None] = mapped_column(String(256))
    match_service: Mapped[str | None] = mapped_column(String(128))
    match_tag_key: Mapped[str | None] = mapped_column(String(128))
    match_tag_value: Mapped[str | None] = mapped_column(String(256))

    assign_person_id: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    assign_department: Mapped[str | None] = mapped_column(String(256))
    assign_cost_center: Mapped[str | None] = mapped_column(String(64))

    enabled: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default=text("1"),
    )
    description: Mapped[str | None] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    created_by_id: Mapped[str | None] = mapped_column(String(36))
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )
    updated_by_id: Mapped[str | None] = mapped_column(String(36))


class CmdbBillingAlertsDedup(Base):
    """Idempotency window for the cloud-billing budget-alert handler.

    UNIQUE on ``(integration_id, month)`` — at most one alert dedup row
    per (cloud account, calendar month). The ``billing_alerts``
    handler consults this row before posting a Feishu alert; the
    decision matrix is:

    * No row exists → send an alert, INSERT this row.
    * ``last_alerted_at`` < 24h ago → skip (silent).
    * ``last_alerted_at`` ≥ 24h ago AND total has grown ≥ 10% over
      ``last_alerted_amount_usd`` → resend, UPDATE this row.
    * ``last_alerted_at`` ≥ 24h ago AND no material increase → skip.

    Stops the daily cmdb billing sync from spamming the alert channel
    when an account is sustained over budget.

    ``integration_id`` is stored as VARCHAR(36) (no FK) on purpose —
    TiDB rejects CHAR↔VARCHAR FK pairings that local MySQL accepts
    (see ``CLAUDE.md`` memory ``feedback_tidb_fk_strictness``). The
    dedup row is essentially a cache; orphan rows after an integration
    delete are harmless and are pruned by month rollover.
    """

    __tablename__ = "cmdb_billing_alerts_dedup"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index(
            "uq_cmdb_billing_alerts_dedup_integration_month",
            "integration_id", "month",
            unique=True,
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[str] = mapped_column(String(36), nullable=False)
    # ``YYYY-MM`` calendar month string — matches the date passed to
    # ``maybe_send_budget_alert`` formatted with ``%Y-%m``.
    month: Mapped[str] = mapped_column(String(7), nullable=False)
    last_alerted_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    last_alerted_amount_usd: Mapped[Decimal] = mapped_column(
        Numeric(14, 2), nullable=False,
    )
    last_alerted_budget_usd: Mapped[Decimal] = mapped_column(
        Numeric(14, 2), nullable=False,
    )


# ── Cloud scope alert dedup (claim-first, replaces alert_rules JSON) ──
#
# rule_type constants — string values land in
# ``cmdb_cloud_scope_alert_fired.rule_type``. ``monthly`` / ``daily``
# back per-rule threshold alerts evaluated by
# :func:`evaluate_scope_alert_rules`; ``daily_spike`` backs the 7-day-
# average comparison done by :func:`_alert_spike`.
CLOUD_SCOPE_ALERT_RULE_TYPE_MONTHLY = "monthly"
CLOUD_SCOPE_ALERT_RULE_TYPE_DAILY = "daily"
CLOUD_SCOPE_ALERT_RULE_TYPE_DAILY_SPIKE = "daily_spike"


class CmdbCloudScopeAlertFired(Base):
    """Claim-first dedup slot for cloud-scope alerts.

    UNIQUE on ``(scope_id, rule_type, period_key, threshold_usd)``.
    Replaces the prior ``cmdb_cloud_scopes.alert_rules[*].last_fired``
    JSON map which was only persisted via the outer cron commit —
    any exception between dispatch and commit would roll back the
    map while leaving the Feishu card in flight, causing the next
    cron tick to re-fire. Same TOCTOU class as the apikey budget
    alerts that PR #298 solved with the same pattern.

    The handler claims a slot BEFORE dispatching: an
    ``INSERT IGNORE``-style write inside ``begin_nested()`` either
    succeeds (caller owns the alert for this period) or raises
    :class:`IntegrityError` (someone else already fired — skip).

    ``period_key`` is ``YYYY-MM`` for monthly rules and
    ``YYYY-MM-DD`` for daily / daily_spike. ``threshold_usd`` is the
    rule's literal threshold for monthly / daily, and the spike
    multiplier (e.g. ``1.5``) for daily_spike — captured for the
    audit trail (which threshold actually triggered the alert), not
    as a dedup discriminator.

    ``rule_type`` carries the evaluated **cost basis** as a suffix when
    that basis is 折前 (``monthly:list`` / ``daily:list``); 实付 keeps the
    bare value (``monthly`` / ``daily``). Suffixing only the list side is
    what lets the discriminator widen with no migration and no backfill —
    a row claimed before 2026-07-29 keeps deduping the net question it was
    claimed for. ``daily_spike`` never carries a suffix: exactly one spike
    comparison runs per scope per day and its basis is picked by data
    completeness, so two bases cannot both fire. See
    :func:`alert_slot_rule_type`.

    **At-most-one-per-period-per-basis guarantee**: while the UNIQUE
    constraint includes ``threshold_usd`` (so multiple historical rows
    under different thresholds can coexist in the table for audit
    archaeology), the caller path
    (:func:`evaluate_scope_alert_rules` /
    :func:`sync_billing_daily_for_integration`) consults
    :func:`_period_already_alerted` BEFORE attempting a claim — that
    helper checks existence on ``(scope_id, rule_type, period_key)``
    only, where ``rule_type`` is the basis-suffixed discriminator above.
    An admin tuning a threshold mid-period (e.g. monthly budget $500 →
    $100, or daily-spike multiplier 1.5× → 1.2×) does NOT re-page the
    same scope on the same data; the operator-facing semantics are "one
    alert per scope per period **per cost basis**, regardless of
    knob-tuning". The basis belongs in the key because a 折前 threshold
    and a 实付 threshold are different questions about different numbers
    — before 2026-07-29 a net monthly alert firing on the 3rd silently
    suppressed the list monthly alert for the rest of the month, and list
    is the engineering-facing default.

    See migration 171.
    """

    __tablename__ = "cmdb_cloud_scope_alert_fired"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        UniqueConstraint(
            "scope_id", "rule_type", "period_key", "threshold_usd",
            name="uq_cmdb_cloud_scope_alert_fired",
        ),
        Index(
            "ix_cmdb_cloud_scope_alert_fired_period",
            "period_key", "rule_type",
        ),
        Index(
            "ix_cmdb_cloud_scope_alert_fired_scope_fired",
            "scope_id", "fired_at",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    scope_id: Mapped[str] = mapped_column(String(36), nullable=False)
    rule_type: Mapped[str] = mapped_column(String(32), nullable=False)
    period_key: Mapped[str] = mapped_column(String(16), nullable=False)
    threshold_usd: Mapped[Decimal] = mapped_column(
        Numeric(14, 4), nullable=False, default=Decimal("0"),
    )
    fired_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    current_usd: Mapped[Decimal] = mapped_column(
        Numeric(14, 4), nullable=False, default=Decimal("0"),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )


# Owner-governance event kinds. One table stores all of them — the
# detail-page "Owner 与治理记录" tab fetches by scope_id and renders a
# chronological timeline. Crons create the ``_pending`` rows; the
# owner-click endpoint flips them to leaf kinds (confirmed /
# transferred / keep / cleanup / handover_it). 14-day-no-response
# escalation writes a SEPARATE ``escalated_to_it`` row (not a status
# flip on the pending row) so cause→consequence stays threaded via
# ``parent_confirmation_id``.
OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING = "semiannual_pending"
OWNER_CONFIRMATION_KIND_SEMIANNUAL_CONFIRMED = "semiannual_confirmed"
OWNER_CONFIRMATION_KIND_SEMIANNUAL_TRANSFERRED = "semiannual_transferred"
OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING = "zero_spend_pending"
OWNER_CONFIRMATION_KIND_ZERO_SPEND_KEEP = "zero_spend_keep"
OWNER_CONFIRMATION_KIND_ZERO_SPEND_CLEANUP = "zero_spend_cleanup"
OWNER_CONFIRMATION_KIND_ZERO_SPEND_HANDOVER_IT = "zero_spend_handover_it"
# Historical only since 2026-08-24 (migration 531): IT escalation was
# retired in favour of re-sending the prompt to the OWNER daily. No code
# writes this kind any more; the constant stays so the timeline keeps
# rendering the rows already in the table.
OWNER_CONFIRMATION_KIND_ESCALATED_TO_IT = "escalated_to_it"
# Standalone owner-transfer event — every ``assign_scope_owner`` call
# that's NOT closing an outstanding ``semiannual_pending`` row writes
# one of these so the governance timeline always reflects "this scope
# changed owners on date D, from A to B, by actor C." Without this an
# admin-initiated rebind or an owner self-transfer (with no pending
# prompt outstanding) leaves only an ``audit_log`` row, which the
# detail-page timeline does not read — owners and auditors would miss
# the transfer entirely.
OWNER_CONFIRMATION_KIND_OWNER_TRANSFERRED = "owner_transferred"
# Weekly idle-scope sweep (``cmdb.scope_idle_sweep``)留痕 — every time
# the sweep notifies an owner/admin that a scope has 0 active grants AND
# $0 spend over the trailing 30 days, it writes ONE of these per
# notified scope so the detail-page governance timeline shows "system
# flagged this scope as idle on date D and pinged X". Written as a leaf
# event (status='responded') — it's a notification record, not an
# action the owner must answer, so it is NOT picked up by
# the daily reminder sweep (which only re-sends the two ``_pending`` kinds).
# Dedup is inherited from the sweep itself (``last_idle_notified_at`` +
# IDLE_SWEEP_DEDUP_DAYS), so at most one row per scope per ~14 days.
OWNER_CONFIRMATION_KIND_IDLE_NOTIFIED = "idle_notified"

# ── Read-time-only kinds ────────────────────────────────────────────
# These two are NEVER persisted to
# ``cmdb_cloud_scope_owner_confirmations``. They are synthesized from
# ``audit_logs`` by ``OwnerConfirmationService.list_timeline_for_scope``
# so the governance timeline shows the COMPLETE owner history of a
# scope, not just the part that happened to flow through
# ``assign_scope_owner``. Two owner-change paths never reached the
# leaf-row writer:
#
#   * ``cloud.scope.auto_owner`` — the sync's auto-claim picks the
#     FIRST owner of a freshly discovered scope. ``assign_scope_owner``
#     deliberately skips the ``owner_transferred`` leaf when there was
#     no previous owner (it would render as "transferred from —"), so
#     the initial assignment was invisible on the timeline.
#   * ``cmdb.cloud_scope.owner_force_transferred`` — offboarding's
#     force-transfer rewrites ``scope.owner_person_id`` directly in
#     ``hr_lifecycle/advance_service.py``; it never calls
#     ``assign_scope_owner`` at all.
#
# Synthesizing at read time (rather than back-writing rows + a backfill
# migration) keeps exactly ONE source of truth per event and yields the
# full pre-existing history for free — 212 such audit rows in prod as
# of 2026-08-24, so the extra indexed SELECT is cheap.
OWNER_CONFIRMATION_KIND_OWNER_AUTO_ASSIGNED = "owner_auto_assigned"
# ``cloud.scope.assign_owner`` — the canonical owner-change path
# (admin re-bind, owner self-transfer, the transfer dialog). 138 of
# these existed in production on 2026-08-24 and NONE of them were
# visible on the timeline: the stored ``owner_transferred`` leaf is
# only written when there is no outstanding prompt to close, so every
# transfer that happened to answer a pending card left no row of its
# own. The derived event also carries the CORRECT predecessor and the
# acting user, which the stored rows do not.
OWNER_CONFIRMATION_KIND_OWNER_ASSIGNED = "owner_assigned"
OWNER_CONFIRMATION_KIND_OWNER_FORCE_TRANSFERRED = "owner_force_transferred"

# Response kind (never a row ``kind``) stamped on a ``_pending`` row
# that is closed WITHOUT an owner answer because the scope changed
# hands while the prompt was outstanding. The card was addressed to
# the previous owner, so neither confirming nor escalating it is
# meaningful — the prompt is void and the semiannual clock must be
# free to re-fire for the NEW owner.
#
# Before this existed, a force-transfer left the pending row untouched:
# ``list_scopes_due_semiannual`` skips scopes that already have an open
# pending row, so the new owner was NEVER prompted, and the daily
# escalation sweep kept paging IT about a card sent to someone who had
# left the company (7 such rows in prod on 2026-08-24, oldest from
# 2026-05-19).
OWNER_CONFIRMATION_RESPONSE_VOIDED_OWNER_CHANGED = "voided_owner_changed"

# Response kind (never a row ``kind``) for a prompt closed because the
# subscription it asks about is gone — hard-deleted, or decommissioned
# (``status='closed'``; decommission deliberately does NOT set
# ``deleted_at``, billing FKs retain the row). Asking its owner to
# confirm ownership of a closed account is a question with no answer,
# and under the daily-reminder cadence it would be asked every morning.
OWNER_CONFIRMATION_RESPONSE_VOIDED_SCOPE_CLOSED = "voided_scope_closed"

# A ``_pending`` row tracks status; leaf-kind rows are written with
# status='responded' right away.
OWNER_CONFIRMATION_STATUS_PENDING = "pending"
OWNER_CONFIRMATION_STATUS_RESPONDED = "responded"
# Historical only (migration 531 re-opened every such row to
# ``pending``). It was a dead end: the respond endpoints accept only a
# PENDING row, so an owner could never answer an escalated prompt and
# nothing would ask them again.
OWNER_CONFIRMATION_STATUS_ESCALATED = "escalated"


class CmdbCloudScopeOwnerConfirmation(Base):
    """Persistent timeline of owner-governance events on a cloud scope.

    Read by the detail-page "Owner 与治理记录" tab to render a unified
    history; written by the two new crons and the owner-respond
    endpoints. See the kind/status constants above for the event taxonomy.

    Snapshot columns capture the state shown to the owner at the moment
    of the prompt (spend / member count) so the timeline UI can show
    "what you saw when you decided" — independent of later cost-facts
    refreshes or membership changes.
    """

    __tablename__ = "cmdb_cloud_scope_owner_confirmations"
    __table_args__ = (
        Index("ix_cmdb_oc_scope_created", "scope_id", "created_at"),
        Index("ix_cmdb_oc_kind_status", "kind", "status"),
        Index("ix_cmdb_oc_owner_status", "owner_person_id", "status"),
    )

    # NOTE: scope_id / *_person_id columns intentionally have NO DB-level
    # FK constraint — prod TiDB and local Docker MySQL have drifted on
    # the target columns' type + collation (cmdb_cloud_scopes.id is
    # VARCHAR(36)/utf8mb4_bin on prod, identity_persons.id is
    # CHAR(36)/utf8mb4_unicode_ci on prod, both are VARCHAR(36)/default
    # locally), so a single ``ForeignKey`` declaration can't satisfy both
    # environments. Referential integrity is enforced by the service
    # layer (``OwnerConfirmationService`` validates scope + person
    # existence on every insert) + soft-delete semantics on scopes
    # (decommission writes ``deleted_at``, never DELETEs). See migration
    # 168's top docstring for the full TiDB-FK-strictness story.
    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    scope_id: Mapped[str] = mapped_column(String(36), nullable=False)
    owner_person_id: Mapped[str | None] = mapped_column(String(36))
    kind: Mapped[str] = mapped_column(String(64), nullable=False)
    status: Mapped[str] = mapped_column(
        String(32),
        nullable=False,
        default=OWNER_CONFIRMATION_STATUS_PENDING,
        server_default=OWNER_CONFIRMATION_STATUS_PENDING,
    )
    response_kind: Mapped[str | None] = mapped_column(String(64))
    response_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    responded_by_person_id: Mapped[str | None] = mapped_column(String(36))
    transferred_to_person_id: Mapped[str | None] = mapped_column(String(36))
    reason: Mapped[str | None] = mapped_column(Text)
    spend_recent_30d_usd: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    spend_recent_180d_usd: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    spend_last_month_usd: Mapped[Decimal | None] = mapped_column(Numeric(14, 6))
    member_count_snapshot: Mapped[int | None] = mapped_column(Integer)
    # When the prompt was FIRST sent. Never bumped by a reminder — the
    # timeline's "how long has this been open" answer depends on it.
    notified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    notification_outbox_id: Mapped[str | None] = mapped_column(String(36))
    # Daily-reminder bookkeeping (migration 531). The cron re-sends the
    # same card to the OWNER once a day until they answer; these two
    # columns are its at-most-once-per-day guard and the "已提醒 N 次"
    # the timeline renders.
    last_reminded_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    reminder_count: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default="0", default=0,
    )
    # Historical only (migration 531 retired IT escalation). Kept so the
    # timeline can still render the rows written before the policy
    # change; nothing sets it any more.
    escalated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    parent_confirmation_id: Mapped[str | None] = mapped_column(String(36))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
        server_default=func.now(), onupdate=func.now(),
    )


# ── cmdb_server_owner_confirmations (migration 214) ───────────────
#
# Per-owner, per-half-year inventory pass on cloud servers. Differs
# from CmdbCloudScopeOwnerConfirmation in two ways:
#   1. Grouped by OWNER, not by individual server. A single card
#      lists every due server an owner holds so we don't spam
#      someone holding 30 VMs with 30 separate cards.
#   2. Snapshot is JSON-encoded list of server UUIDs (``server_ids``),
#      not per-resource cost numbers. The card just has to remind
#      the owner the machines exist; spend/utilization decisions
#      live elsewhere (CostAnalysisPage).
#
# Lifecycle:
#  pending  — cron created the row, sent the Feishu card
#  responded — owner clicked "确认仍在使用" (in-app or via card CTA);
#              every ``server_ids`` element gets ``last_owner_confirmed_at``
#              bumped to NOW so the next cron tick skips them for ~180d
#  escalated — 14 days passed with no response (cron transitions this
#              and re-notifies IT; phase-2 follow-up, not in v1)
#  auto_resolved — every snapshot server became ineligible AFTER the
#              card went out (stopped / decommissioned / soft-deleted /
#              re-assigned / vanished in a re-sync). The reconcile cron
#              closes the row without owner action — asking someone to
#              confirm a machine that no longer exists (or is already
#              addressed by the long-idle surface) is pure noise, and a
#              stuck pending row also blocks the re-prompt interlock so
#              the owner would never get another半年盘点 card. No
#              ``last_owner_confirmed_at`` stamp — if a stopped machine
#              comes back it becomes due again on the next scan.
#
# Dedup: UNIQUE(owner_person_id, period_start). The cron computes
# period_start as today's date; running the cron twice on the same
# day for the same owner is a no-op.


# Status constants for cmdb_server_owner_confirmations.status.
SERVER_OWNER_CONFIRMATION_STATUS_PENDING = "pending"
SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED = "responded"
SERVER_OWNER_CONFIRMATION_STATUS_ESCALATED = "escalated"
SERVER_OWNER_CONFIRMATION_STATUS_AUTO_RESOLVED = "auto_resolved"

# Response kind — confirmed_all (most common; one click bumps every
# server) vs partial (owner used the per-server actions on some
# subset). Partial is computed by the response endpoint, not surfaced
# as a CTA on the card.
SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL = "confirmed_all"
SERVER_OWNER_CONFIRMATION_RESPONSE_PARTIAL = "partial"


class CmdbServerOwnerConfirmation(Base):
    """Per-owner half-yearly cloud-server inventory confirmation.

    See module docstring above for the lifecycle. Written by
    ``ServerOwnerConfirmationService`` (cron path + owner-response
    endpoint). Read by ``GET /cmdb/server-confirmations/mine`` so the
    UI can list pending tasks for the signed-in owner.
    """

    __tablename__ = "cmdb_server_owner_confirmations"
    __table_args__ = (
        # Idempotency anchor: one row per owner per cron tick (the
        # cron computes period_start as today's date, so re-runs land
        # on the same key and ON CONFLICT shorts-circuits). Matches
        # the (scope_id, kind, status='pending') no-duplicate-pending
        # guard the scope service uses, but tighter — we don't need
        # the kind axis here since there's only one card type.
        UniqueConstraint(
            "owner_person_id", "period_start",
            name="uq_cmdb_server_owner_conf_owner_period",
        ),
        Index("ix_cmdb_server_owner_conf_owner_status", "owner_person_id", "status"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Owner the card was sent to. Like CmdbCloudScopeOwnerConfirmation
    # there is NO DB-level FK (see that table's docstring for why —
    # TiDB-strictness on cross-collation FKs). The service layer
    # validates owner existence and skips leavers per the skip rules.
    owner_person_id: Mapped[str] = mapped_column(String(36), nullable=False)
    # Date the card was issued (NOT a timestamp — the cron only fires
    # once a day, so date precision keeps the unique index small and
    # gives the natural "one-per-period" semantics).
    period_start: Mapped[date_type] = mapped_column(Date, nullable=False)
    # JSON array of server UUID strings. Snapshot of what was due
    # when the card went out; the owner's "确认全部" applies only to
    # this snapshot so an admin re-assigning a server mid-period
    # doesn't get a stale confirmation.
    server_ids: Mapped[list[str]] = mapped_column(JSON, nullable=False)
    status: Mapped[str] = mapped_column(
        String(32),
        nullable=False,
        default=SERVER_OWNER_CONFIRMATION_STATUS_PENDING,
        server_default=SERVER_OWNER_CONFIRMATION_STATUS_PENDING,
    )
    response_kind: Mapped[str | None] = mapped_column(String(32))
    responded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    notified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    notification_outbox_id: Mapped[str | None] = mapped_column(String(36))
    escalated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
        server_default=func.now(), onupdate=func.now(),
    )


# ── cmdb_resource_creators (migration 198) ────────────────────────
#
# Per-resource attribution cache populated from cloud audit logs.
# See migration 198 docstring + ``cmdb_resource_creator_service.py``
# for the full lifecycle (on-demand resolve + cron sweep).

RESOURCE_CREATOR_STATUS_PENDING = "pending"
RESOURCE_CREATOR_STATUS_RESOLVED = "resolved"
RESOURCE_CREATOR_STATUS_NOT_FOUND = "not_found"
RESOURCE_CREATOR_STATUS_WINDOW_EXCEEDED = "window_exceeded"
RESOURCE_CREATOR_STATUS_ERROR = "error"
RESOURCE_CREATOR_STATUS_UNSUPPORTED = "unsupported"
RESOURCE_CREATOR_STATUSES = frozenset({
    RESOURCE_CREATOR_STATUS_PENDING,
    RESOURCE_CREATOR_STATUS_RESOLVED,
    RESOURCE_CREATOR_STATUS_NOT_FOUND,
    RESOURCE_CREATOR_STATUS_WINDOW_EXCEEDED,
    RESOURCE_CREATOR_STATUS_ERROR,
    RESOURCE_CREATOR_STATUS_UNSUPPORTED,
})

# Status values that won't be retried by the cron sweep —
# ``resolved`` is the success case, the other two are permanent
# misses (resource pre-dates the audit retention window OR the
# driver doesn't support audit-log lookup for this provider).
# ``not_found`` is also terminal in practice (within the retention
# window, the audit log doesn't lie); the sweep treats it as
# terminal but the on-demand surface re-attempts on user click in
# case the provider had a transient empty response.
RESOURCE_CREATOR_TERMINAL_STATUSES = frozenset({
    RESOURCE_CREATOR_STATUS_RESOLVED,
    RESOURCE_CREATOR_STATUS_NOT_FOUND,
    RESOURCE_CREATOR_STATUS_WINDOW_EXCEEDED,
    RESOURCE_CREATOR_STATUS_UNSUPPORTED,
})

RESOURCE_CREATOR_SOURCE_CLOUDTRAIL = "cloudtrail"
RESOURCE_CREATOR_SOURCE_GCP_AUDIT = "gcp_audit"
RESOURCE_CREATOR_SOURCE_ALIYUN_ACTIONTRAIL = "aliyun_actiontrail"
RESOURCE_CREATOR_SOURCE_AZURE_ACTIVITY = "azure_activity"
RESOURCE_CREATOR_SOURCES = frozenset({
    RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
    RESOURCE_CREATOR_SOURCE_GCP_AUDIT,
    RESOURCE_CREATOR_SOURCE_ALIYUN_ACTIONTRAIL,
    RESOURCE_CREATOR_SOURCE_AZURE_ACTIVITY,
})

# Hard cap on retries — beyond this the cron stops attempting so a
# perma-broken provider config doesn't burn through audit-API rate
# budget every cron tick. Operator can force-retry via
# ``POST /api/v1/cmdb/cost/resources/{integration_id}/{resource_id}/creator/retry``
# (handler: ``resource_creator_retry_endpoint``) which zeroes the
# attempt count + clears the cooldown + fires a fresh resolve.
RESOURCE_CREATOR_MAX_ATTEMPTS = 5


class CmdbResourceCreator(Base):
    """Per-resource "who created this" attribution cached from the
    cloud provider's audit log.

    Populated by :func:`cmdb_resource_creator_service.resolve_resource_creator`
    (on-demand, when a user opens a resource drawer or calls the
    HTTP endpoint) and the ``cmdb.resource_creator_sweep`` cron (
    pre-resolves rows for newly-discovered resources seen in
    ``cmdb_cost_facts`` so the dashboard is never the first hit).

    Schema details + index strategy live in migration 198's
    top-of-file docstring; this class is the SQLAlchemy mirror.
    """

    __tablename__ = "cmdb_resource_creators"
    __table_args__ = (
        UniqueConstraint(
            "integration_id", "resource_id",
            name="uq_resource_creators_integ_rid",
        ),
        Index(
            "ix_resource_creators_sweep",
            "status", "last_attempt_at",
        ),
        Index(
            "ix_resource_creators_person",
            "creator_person_id", "resolved_at",
        ),
        Index("ix_resource_creators_email", "creator_email"),
    )

    id: Mapped[int] = mapped_column(
        BigInteger, primary_key=True, autoincrement=True,
    )
    integration_id: Mapped[str] = mapped_column(
        String(36),
        ForeignKey("integrations.id", ondelete="RESTRICT"),
        nullable=False,
    )
    resource_id: Mapped[str] = mapped_column(String(512), nullable=False)
    resource_type: Mapped[str | None] = mapped_column(String(128))
    creator_principal_id: Mapped[str | None] = mapped_column(String(512))
    creator_email: Mapped[str | None] = mapped_column(String(255))
    creator_display_name: Mapped[str | None] = mapped_column(String(255))
    creator_person_id: Mapped[str | None] = mapped_column(
        String(36),
        ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    created_at_cloud: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    resolved_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    source: Mapped[str] = mapped_column(String(32), nullable=False)
    status: Mapped[str] = mapped_column(
        String(16),
        nullable=False,
        default=RESOURCE_CREATOR_STATUS_PENDING,
        server_default=RESOURCE_CREATOR_STATUS_PENDING,
    )
    attempt_count: Mapped[int] = mapped_column(
        Integer, nullable=False,
        default=0, server_default=text("0"),
    )
    last_attempt_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    last_error: Mapped[str | None] = mapped_column(Text)
    raw: Mapped[dict | None] = mapped_column(JSON)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )


class CmdbDedicatedLine(Base):
    """专线 (dedicated/leased line) — a manually-maintained register of the
    inter-office / VPN network circuits the IT team contracts from carriers.

    Unlike the cloud-resource tables in this module, these rows are NOT synced
    from any provider — there is no upstream API for a telco leased line. The
    data lives entirely in Helix and is curated by network admins
    (``can_manage_cloud``). The initial population came from the IT operations
    cockpit "网络专线统计" spreadsheet (seeded in migration 310).

    Cost is stored as a plain monthly + annual amount in ``currency`` (CNY for
    every seeded row); no FX is applied at rest. There is no owner FK — a
    leased line belongs to an office/use-case, not a person — only modifier
    tracking for audit (last_modified_by_person_id / last_modified_at).
    """

    __tablename__ = "cmdb_dedicated_lines"

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # 使用位置 — office city or logical endpoint (e.g. "飞连VPN+金山云").
    location: Mapped[str] = mapped_column(String(128), nullable=False)
    # 签约主体 — the legal entity that signs the contract.
    contracting_entity: Mapped[str | None] = mapped_column(String(255))
    # 运营商 — carrier / ISP (中国电信 / 中国联通 / 世纪互联 …).
    carrier: Mapped[str | None] = mapped_column(String(128))
    # 付款方式 — free text (月付 / 年付 / 年付【积分兑换】 …).
    payment_method: Mapped[str | None] = mapped_column(String(64))
    # 每月/元 + 年度/元 — stored in ``currency`` (CNY for all seeded rows).
    monthly_cost: Mapped[Decimal | None] = mapped_column(Numeric(14, 2))
    annual_cost: Mapped[Decimal | None] = mapped_column(Numeric(14, 2))
    currency: Mapped[str] = mapped_column(
        String(8), nullable=False, default="CNY", server_default="CNY",
    )
    # 续费日期 — recurring "MM-DD" the circuit renews on each year (year-
    # agnostic; e.g. "04-01"). Normalised from the original free text
    # ("每年4月份") in migration 311. Stored as a short string rather than a
    # DATE because the renewal is an annual recurrence, not a calendar date.
    renewal_date: Mapped[str | None] = mapped_column(String(128))
    # 用途 — purpose / scenario description.
    purpose: Mapped[str | None] = mapped_column(Text)
    remark: Mapped[str | None] = mapped_column(Text)

    last_modified_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    last_modified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )


# ── Leaver cloud-IAM residue ─────────────────────────────────────────

# One governance row per (departed person × cloud IAM user that still
# carries access). See ``leaver_iam_residue.py`` for why the check is
# framed against the provider mirror rather than against Helix's own
# grant rows.
RESIDUE_STATUS_OPEN = "open"
RESIDUE_STATUS_RESOLVED = "resolved"
RESIDUE_STATUS_DISMISSED = "dismissed"

# ``resolution`` values on a resolved row.
#   cleared  — the residue stopped appearing in the mirror (someone
#              removed it, here or in the console)
#   revoked  — an admin ran the cleanup chain from Helix
RESIDUE_RESOLUTION_CLEARED = "cleared"
RESIDUE_RESOLUTION_REVOKED = "revoked"
# The row stopped being detected WITHOUT the access going away. Keeping
# these distinct from ``cleared`` matters: ``cleared`` asserts the
# cloud-side permission is gone, and neither of these verified that.
#   reclassified  — the IAM user was declared a service account
#   not_a_leaver  — the person is active again (rehire / HRIS flip-back)
RESIDUE_RESOLUTION_RECLASSIFIED = "reclassified"
RESIDUE_RESOLUTION_NOT_A_LEAVER = "not_a_leaver"


class CmdbLeaverIamResidue(Base):
    """A departed person's still-live cloud IAM access.

    Exists as a table rather than a live query for three reasons that all
    matter operationally:

    * **Age.** ``first_detected_at`` is the clock the admin page ages
      against — "this leaver has kept production access for 71 days" is
      the number that drives action, and it cannot be recomputed after
      the fact from a live scan.
    * **Dedup.** The alert fires only for rows opened on this run; a live
      query would re-notify every night until someone acted, which is how
      a governance channel gets muted.
    * **Disposition.** ``dismissed`` carries an actor and a reason, so a
      known-good exception (a shared account being re-owned) stops
      nagging without disappearing from the audit trail.
    """

    __tablename__ = "cmdb_leaver_iam_residue"
    __table_args__ = (
        UniqueConstraint(
            "iam_user_id", "person_id",
            name="uq_leaver_iam_residue_user_person",
        ),
        Index("ix_leaver_iam_residue_status", "status"),
    )

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    iam_user_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("cmdb_iam_users.id", ondelete="CASCADE"),
        nullable=False,
    )
    person_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="CASCADE"),
        nullable=False,
    )
    integration_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("integrations.id"), nullable=False,
    )
    # Denormalised so a resolved row still reads correctly after the IAM
    # mirror row is swept away by the provider sync.
    user_name: Mapped[str] = mapped_column(String(255), nullable=False)
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    policy_count: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default="0",
    )
    group_count: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default="0",
    )
    # Snapshot of the last detection: policy_names / group_names /
    # match_basis / mirror_stale. Kept as JSON because it is evidence for
    # a human decision, not a queried dimension.
    detail: Mapped[dict | None] = mapped_column(JSON)
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, server_default=RESIDUE_STATUS_OPEN,
    )
    first_detected_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    last_detected_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    resolution: Mapped[str | None] = mapped_column(String(24))
    resolved_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    resolved_by_person_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    resolution_note: Mapped[str | None] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(),
        onupdate=func.now(), nullable=False,
    )