import uuid
from datetime import UTC, datetime
from datetime import date as date_type
from decimal import Decimal
from typing import Any

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

from app.core.database import Base
from app.core.sql_types import GUID as Uuid

# Statuses that indicate the asset currently has a holder or is in
# flight. Promoted from ``AssetService._IN_USE_STATUSES`` to module
# scope (Round 4 audit fix-up 2026-05-04) so cross-domain consumers
# (offboarding ``_rescan``, governance reports) can use the canonical
# set instead of duplicating it. ``borrowed_by`` carries the borrower
# while status='borrowed'; ``assigned_to`` carries the original owner
# (or NULL for borrow-from-idle).
ASSET_IN_USE_STATUSES: frozenset[str] = frozenset({
    "assigned",
    "claimed",
    "borrowed",
    "claim_pending",
    "return_pending",
    "borrow_pending",
    "borrow_return_pending",
    "transferring",
})

# Non-standard but real-world state: the holder has left the company and
# the asset was not physically returned (employment dispute pending,
# overseas employee carried the laptop, IT consciously chose to leave the
# asset attributed to the ex-employee). Set by offboarding ``skip_blocker``
# on a hardware blocker OR by an admin via the asset detail page. The
# asset is still company property; ``assigned_to`` is preserved so the
# audit trail and "last holder" attribution are intact. Recovery flips
# the asset back to ``idle`` via ``AssetService.recover_from_leaver``.
#
# Deliberately NOT in ``ASSET_IN_USE_STATUSES`` — the offboarding scanner
# uses that set to decide whether the leaver still holds anything, and a
# ``held_by_leaver`` row by construction belongs to an *already-departed*
# person, so re-detecting it on a rehire would re-open the process for
# the wrong reason.
ASSET_STATUS_HELD_BY_LEAVER = "held_by_leaver"

# Office-held mode (migration 183, 2026-05-19). Companion to
# ``holder_kind='office'`` — the device physically lives at a city office
# (meeting-room kit, shared dock, spare device on a shelf) with no Person
# assignee. ``assigned_to`` MUST be NULL; ``holder_office_id`` points at
# the responsible ``asset_office_addresses`` row, and ``OfficeAddress
# .admin_person_id`` is who receives inventory pings + offboarding
# handoffs. Pre-183 these assets were silently routed by the inventory
# classifier via a free-text ``location`` match — that worked for idle
# rows but broke the moment T+0 force_transfer reattributed a leaver's
# laptop to whatever active person caught the toss, hiding it from the
# office-routing path. Making the mode explicit also makes "recycle to
# office" a first-class operation (rather than "unassign + hope").
#
# Like ``held_by_leaver`` this status is deliberately NOT in
# ``ASSET_IN_USE_STATUSES``: the offboarding scanner treats this row as
# "nobody's personal device, leave it alone". The lifecycle ends with
# either ``claim_asset`` (someone picks it up, → personal/claimed) or
# ``dispose_asset``.
ASSET_STATUS_HELD_BY_OFFICE = "held_by_office"

# ── purchase_cost provenance (migration 604, 2026-09-05) ──────────────
# An operator-recorded amount. Every row predating migration 604 is
# backfilled to this, which is honest: the amounts that WERE present came
# from intake or a bulk update.
PURCHASE_COST_SOURCE_RECORDED = "recorded"
# Median of same-model devices that have a price. An estimate, and the UI
# must say so — it exists to make depreciation possible on the 2026-04
# legacy import, not to make the completeness metric look better.
PURCHASE_COST_SOURCE_ESTIMATED_MODEL = "estimated_model"
# Matched against the finance fixed-asset register by serial number. The
# authoritative source; overwrites an estimate, never the reverse.
PURCHASE_COST_SOURCE_FINANCE_REGISTER = "finance_register"
PURCHASE_COST_SOURCES = (
    PURCHASE_COST_SOURCE_RECORDED,
    PURCHASE_COST_SOURCE_ESTIMATED_MODEL,
    PURCHASE_COST_SOURCE_FINANCE_REGISTER,
)

ASSET_HOLDER_KIND_PERSONAL = "personal"
ASSET_HOLDER_KIND_OFFICE = "office"

# ---------------------------------------------------------------------------
# Canonical ``asset_items.status`` vocabulary — THE list (2026-08-07).
#
# Ordered as the lifecycle reads, and enforced at write time by
# ``AssetItem.validate_status`` below. ``schemas.ALLOWED_STATUSES`` is
# derived from this tuple so the request validator and the ORM can never
# drift apart.
#
# Why the write-time guard exists: ``in_stock`` (a pre-Helix Excel-import
# synonym for idle) sat in one prod row for months, frozen — every
# holder-changing gate whitelists canonical statuses, so the asset could
# not be dispatched to anyone and 「变更持有」 just refused. Nothing in the
# code wrote that token; it arrived through a bulk import that bypassed
# the request schema. Migration 454 retired it from the data, and this
# validator is what makes the next such import fail loudly at the write
# instead of silently freezing an asset. See PLATFORM.md.
#
# Retired from the vocabulary in the same pass (0 rows in prod, 0 writers,
# never reachable): ``maintenance`` (repairs live on ``device_status`` —
# ``under_repair`` / ``pending_repair``), ``retired`` (duplicate of
# ``disposed``), ``faulty`` (name-collides with ``device_status='faulty'``,
# which is the real one), ``testing`` / ``legacy`` (import-era leftovers).
ASSET_STATUSES: tuple[str, ...] = (
    # In stock, no holder. ``idle`` is canonical; ``available`` is its
    # legacy twin — still accepted on read, no longer written by any path
    # (``unassign_asset`` was the last writer, 2026-08-07).
    "idle",
    "available",
    # Held by a person. ``claimed`` = employee picked it up;
    # ``assigned`` = admin dispatched it.
    "claimed",
    "assigned",
    "borrowed",
    # Held, but not by a person.
    ASSET_STATUS_HELD_BY_OFFICE,
    ASSET_STATUS_HELD_BY_LEAVER,
    # In-flight two-phase states. No code writes these today; they are the
    # vocabulary the "don't touch, an operation is pending" guards speak
    # (``ASSET_IN_USE_STATUSES``, inventory snapshot exclusions), and the
    # names a future request→confirm flow would reuse.
    "claim_pending",
    "return_pending",
    "borrow_pending",
    "borrow_return_pending",
    "transferring",
    # Terminal.
    "disposal_pending",
    "disposed",
)

ASSET_STATUS_SET: frozenset[str] = frozenset(ASSET_STATUSES)


class AssetItem(Base):
    __tablename__ = "asset_items"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index("ix_asset_items_assigned_to", "assigned_to"),
        Index("ix_asset_items_status", "status"),
        Index("ix_asset_items_deleted_at", "deleted_at"),
        Index("ix_asset_items_borrowed_by", "borrowed_by"),
        Index("ix_asset_items_holder", "holder_kind", "holder_office_id"),
    )

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )
    asset_tag: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
    asset_name: Mapped[str | None] = mapped_column(String(128))
    category: Mapped[str] = mapped_column(String(32), nullable=False)
    # Free-text brand / model strings — kept as the canonical display
    # value for back-compat with reports / list endpoints. New writes
    # also populate ``brand_id`` / ``model_id`` (added in migration 071)
    # so admins can curate the lookup.
    brand: Mapped[str | None] = mapped_column(String(64))
    model: Mapped[str | None] = mapped_column(String(128))
    # FK lookup — populated alongside the text columns by intake / update.
    # Nullable so legacy rows + free-text-only writes still work.
    brand_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("asset_brands.id", ondelete="SET NULL"),
    )
    model_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("asset_models.id", ondelete="SET NULL"),
    )
    serial_number: Mapped[str | None] = mapped_column(String(128), unique=True)
    status: Mapped[str] = mapped_column(String(32), nullable=False, default="idle")
    device_status: Mapped[str] = mapped_column(String(16), nullable=False, default="normal")
    source_type: Mapped[str | None] = mapped_column(String(16))
    order_number: Mapped[str | None] = mapped_column(String(128))
    purchase_date: Mapped[datetime | None] = mapped_column(Date)
    purchase_cost: Mapped[Decimal | None] = mapped_column(Numeric(10, 2))
    # Where ``purchase_cost`` came from (migration 604). One of
    # ``PURCHASE_COST_SOURCES``. Depreciation, BvA and cost attribution
    # all read the amount, so an ESTIMATE has to stay distinguishable
    # from a recorded invoice figure — a single amount column cannot say
    # which it is, and once the two are mixed the number stops being
    # defensible. Read surfaces label ``estimated_model`` explicitly.
    purchase_cost_source: Mapped[str] = mapped_column(
        String(24),
        nullable=False,
        default=PURCHASE_COST_SOURCE_RECORDED,
        server_default=PURCHASE_COST_SOURCE_RECORDED,
    )
    currency: Mapped[str | None] = mapped_column(String(8))
    invoice_number: Mapped[str | None] = mapped_column(String(128))
    warranty_until: Mapped[datetime | None] = mapped_column(Date)
    usage_limit_months: Mapped[int | None] = mapped_column(Integer)
    supplier_name: Mapped[str | None] = mapped_column(String(128))
    supplier_contact: Mapped[str | None] = mapped_column(String(128))
    supplier_phone: Mapped[str | None] = mapped_column(String(64))
    maintenance_note: Mapped[str | None] = mapped_column(Text)
    location: Mapped[str | None] = mapped_column(String(256))

    admin_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )

    assigned_to: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    assigned_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # Office-held mode (migration 183). When ``holder_kind='office'`` the
    # asset has no Person assignee — it lives at the city office pointed
    # at by ``holder_office_id``. Service-layer invariant: ``office``
    # ⇒ assigned_to IS NULL AND holder_office_id IS NOT NULL. The
    # inventory classifier routes these directly via
    # ``holder_office_id`` (no more location string fuzzy-match).
    holder_kind: Mapped[str] = mapped_column(
        String(16),
        nullable=False,
        default="personal",
        server_default="personal",
    )
    holder_office_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("asset_office_addresses.id", ondelete="SET NULL"),
    )

    # Borrow tracking (migration 115). When an asset is loaned to another
    # person, ``borrowed_by`` holds the borrower and ``assigned_to`` is
    # PRESERVED as the original owner — so offboarding scans, "my
    # inventory" listings, and audit trails stay accurate while the asset
    # is out on loan. Cleared on borrow_return. Pre-115 rows had
    # ``assigned_to`` overwritten by the borrower; the migration's
    # backfill recovers the original owner from AssetOperation history.
    borrowed_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    borrow_started_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    # Reservation for an incoming hire (migration 279, 备机闭环). The
    # asset stays in its stock status (idle / available / held_by_office
    # / held_by_leaver) but is earmarked: assign / claim / transfer to
    # anyone OTHER than the reserved person is refused until the
    # reservation is released. Cleared automatically when the asset is
    # assigned to the reserved person.
    reserved_for_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    reserved_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    disposal_type: Mapped[str | None] = mapped_column(String(32))
    disposed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Disposal financials (PR 角色对称/财务闭环). ``disposal_proceeds`` is the
    # cash recovered on sale / lease-return (0 for scrapped / lost /
    # donated). ``disposal_book_value`` is the net book value FROZEN at
    # the disposal date (computed from the depreciation policy at that
    # instant) so the disposal gain/loss (= proceeds − book value) is
    # reproducible even after the policy later changes. ``disposal_currency``
    # mirrors the asset currency at disposal time. All three are also
    # snapshotted into the immutable disposal ``AssetOperation.detail``.
    disposal_proceeds: Mapped[Decimal | None] = mapped_column(Numeric(12, 2))
    disposal_book_value: Mapped[Decimal | None] = mapped_column(Numeric(12, 2))
    disposal_currency: Mapped[str | None] = mapped_column(String(8))
    # S3 object keys for the disposal evidence the admin uploaded with the
    # 处置 action (回收单 / 二手买家付款截图 / 租赁归还回执 / 硬盘销毁照片).
    # Same contract as ``asset_inventory_items.evidence_photos``: the
    # browser PUTs straight to S3 via ``/api/v1/uploads/presign-put`` and
    # only the keys land here; resolve to short-lived presigned GET URLs at
    # render time via ``app.core.storage.presign_get``. Also copied into the
    # immutable disposal ``AssetOperation.detail`` snapshot (migration 591).
    #
    # ``none_as_null=True`` is load-bearing: SQLAlchemy's JSON type renders
    # a Python ``None`` as the JSON literal ``null`` — a real value — not
    # SQL NULL. "This disposal has no evidence" must be SQL NULL, the same
    # thing every pre-591 row already is, or a later
    # ``WHERE disposal_photos IS NOT NULL`` (how many disposals carry
    # proof?) counts rows that carry none. See the 2026-09-02
    # ``cost_facts_raw_purge`` infinite loop for what that costs.
    disposal_photos: Mapped[list[str] | None] = mapped_column(
        JSON(none_as_null=True),
    )

    # Soft-delete (migration 115). ``deleted_at IS NULL`` is the active
    # filter applied to every read. Replaces hard DELETE which used to
    # cascade-delete asset_history / asset_operations and break the
    # audit chain (CLAUDE.md rule 6).
    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    image_urls: Mapped[Any | None] = mapped_column(JSON, default=list)
    custom_fields: Mapped[Any | None] = mapped_column(JSON, default=dict)

    epandian_id: Mapped[int | None] = mapped_column(BigInteger)

    # New columns for Excel import
    owner_company: Mapped[str | None] = mapped_column(String(128))
    usage_company: Mapped[str | None] = mapped_column(String(128))
    expected_return_date: Mapped[date_type | None] = mapped_column(Date)
    repair_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0"))
    repair_total_cost: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False, default=Decimal("0"), server_default=text("0"))
    print_status: Mapped[str | None] = mapped_column(String(16))
    rfid: Mapped[str | None] = mapped_column(String(128))
    purchase_entity: Mapped[str | None] = mapped_column(String(128))

    notes: Mapped[str | None] = mapped_column(Text)

    # Asset inventory (盘点) — denormalised "last confirmed" state so
    # the asset list page can show a "最近盘点" column + filter without
    # joining to ``asset_inventory_items``. Written when an inventory
    # item flips to ``confirmed`` or ``issue_reported``. ``status``
    # mirrors ``asset_inventory_items.status`` for the most recent
    # responded item ('confirmed' | 'issue_reported'); pending items
    # do NOT update these columns.
    last_inventory_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    last_inventory_cycle_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("asset_inventory_cycles.id", ondelete="SET NULL"),
    )
    last_inventory_status: Mapped[str | None] = mapped_column(String(32))

    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(),
    )

    # Relationships
    history: Mapped[list["AssetHistory"]] = relationship(
        back_populates="asset",
        order_by="desc(AssetHistory.created_at)",
    )
    operations: Mapped[list["AssetOperation"]] = relationship(
        back_populates="asset",
        order_by=lambda: (
            AssetOperation.created_at.desc(),
            AssetOperation.operation_number.desc(),
        ),
    )

    @validates("status")
    def validate_status(self, _key: str, value: str) -> str:
        """Refuse any status outside the canonical vocabulary.

        Deliberately at the ORM layer, not (only) in the request schema:
        the token that froze a laptop for months (``in_stock``) came in
        through a bulk import that never touched ``AssetCreateRequest``.
        Every write — service, script, migration-adjacent backfill —
        goes through here. Loading an existing row does NOT trigger this,
        so a not-yet-migrated environment still reads its legacy rows.
        """
        if value not in ASSET_STATUS_SET:
            raise ValueError(
                f"Unknown asset status {value!r}. Allowed: "
                f"{', '.join(ASSET_STATUSES)}"
            )
        return value


class AssetLocation(Base):
    """Managed list of asset locations with codes for tag generation.

    Also the anchor of asset-domain administrative authority: the
    persons listed in ``asset_location_admins`` for a location are the
    ``asset_admin`` role holders scoped to it (migration 535). Before
    that the authority hung off ``asset_office_addresses``, which
    forced every "who covers this asset" question through a
    city/country inference ladder; ``asset_items.location`` is a
    managed, resolver-gated value, so anchoring here makes the lookup
    a direct one.
    """
    __tablename__ = "asset_locations"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )
    code: Mapped[str] = mapped_column(String(16), nullable=False, unique=True)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0"))
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=text("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 AssetLocationAdmin(Base):
    """Many-to-many: which persons administer which asset location.

    Single source of truth for ``asset_admin`` role membership — the
    role is fully auto-derived from this table (see
    ``app.domains.asset.location_admin_sync``) and manual grants are
    rejected with ``ROLE_AUTO_DERIVED``. It is also the region scope:
    an ``asset_admin`` may only mutate assets whose resolved location
    is one they are listed against.

    ``is_primary`` marks the one row per location that single-recipient
    flows address (inventory office-held confirmations, recycle-to-office
    notifications). Enforced by the service layer, not a DB constraint —
    MySQL has no partial unique index, and a location with no primary is
    a legitimate intermediate state while an admin list is being edited.

    Distinct from ``AssetOfficeLocalAdmin``: that one is the office's
    当地 Admin (a local IT contact for person-anchored HR flows) and
    grants no asset authority at all.
    """
    __tablename__ = "asset_location_admins"
    __mapper_args__ = {"eager_defaults": True}

    location_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("asset_locations.id", ondelete="CASCADE"),
        primary_key=True,
    )
    person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id", ondelete="CASCADE"),
        primary_key=True,
    )
    is_primary: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default=text("0"),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )


class AssetCategoryConfig(Base):
    """Managed list of asset categories — the source of truth for the
    category segment of an asset tag (``IT-{entity}-{code}-{ts}``).

    ``asset_category`` is the stable asset-category enum value
    (``laptop`` / ``desktop`` / ...) that ``AssetItem.category`` stores;
    it links an asset to its config row. ``code`` is the literal tag
    segment (``NB`` / ``DT`` / ...) — admins edit it via the 分类管理 tab
    and it takes effect on the next intake.
    """
    __tablename__ = "asset_category_configs"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )
    code: Mapped[str] = mapped_column(String(16), nullable=False, unique=True)
    asset_category: Mapped[str | None] = mapped_column(String(32), unique=True)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    parent_code: Mapped[str | None] = mapped_column(String(16))
    usage_limit_months: Mapped[int | None] = mapped_column(Integer)
    unit: Mapped[str | None] = mapped_column(String(32))
    notes: Mapped[str | None] = mapped_column(Text)
    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0"))
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=text("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 AssetBrand(Base):
    """Managed list of asset brands (Apple, Lenovo, Dell, ...).

    Two-level lookup with :class:`AssetModel` so the intake form can
    show "select Apple → MacBook Pro 14 / Air 13 / ..." and admins can
    curate / merge / deactivate values via the asset settings page.

    ``auto_created=True`` rows were inserted by the intake AutoComplete
    (admin typed a value the lookup didn't have) — surfaces in the
    settings UI with a chip so the admin can decide whether to rename,
    merge, or deactivate.
    """
    __tablename__ = "asset_brands"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    name: Mapped[str] = mapped_column(
        String(64), nullable=False, unique=True,
    )
    display_name: Mapped[str | None] = mapped_column(String(128))
    notes: Mapped[str | None] = mapped_column(Text)
    sort_order: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="1",
    )
    auto_created: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, 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 AssetModel(Base):
    """Managed list of asset models, scoped under a brand.

    Composite uniqueness is on (brand_id, name) so the same model name
    can recur across brands — common for accessory / cable / generic
    SKUs.
    """
    __tablename__ = "asset_models"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    brand_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("asset_brands.id", ondelete="CASCADE"),
        nullable=False,
    )
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    notes: Mapped[str | None] = mapped_column(Text)
    sort_order: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default="1",
    )
    auto_created: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, 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 AssetHistory(Base):
    """Legacy history table kept for backward compatibility."""
    __tablename__ = "asset_history"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )
    asset_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("asset_items.id"),
        nullable=False,
    )
    action: Mapped[str] = mapped_column(String(32), nullable=False)

    from_user: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    to_user: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    operator_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    notes: Mapped[str | None] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        default=lambda: datetime.now(UTC),
        server_default=func.now(),
    )

    # Relationships
    asset: Mapped["AssetItem"] = relationship(back_populates="history")


class AssetOperation(Base):
    """Voucher/order-based operation tracking for asset lifecycle."""
    __tablename__ = "asset_operations"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )
    operation_number: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    asset_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("asset_items.id"),
        nullable=False,
    )
    operation_type: Mapped[str] = mapped_column(String(32), nullable=False)
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")

    from_user_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    to_user_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    operator_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
        nullable=False,
    )
    notes: Mapped[str | None] = mapped_column(Text)
    detail: Mapped[Any | None] = mapped_column(JSON)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        default=lambda: datetime.now(UTC),
        server_default=func.now(),
    )
    confirmed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    confirmed_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )

    # Relationships
    asset: Mapped["AssetItem"] = relationship(back_populates="operations")


class AssetProcurementOrder(Base):
    """Procurement orders for asset purchasing/leasing."""
    __tablename__ = "asset_procurement_orders"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )
    order_number: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    order_type: Mapped[str] = mapped_column(String(16), nullable=False)
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="draft")

    applicant_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
        nullable=False,
    )
    supplier_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("asset_suppliers.id"),
    )
    department: Mapped[str | None] = mapped_column(String(128))
    supplier_name: Mapped[str | None] = mapped_column(String(128))
    total_amount: Mapped[Decimal | None] = mapped_column(Numeric(12, 2))
    items: Mapped[Any | None] = mapped_column(JSON)
    notes: Mapped[str | None] = mapped_column(Text)

    requested_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    approved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    approved_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    completed_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 AssetSoftwareCatalog(Base):
    """Software asset catalog — tracks all SaaS/IT software subscriptions."""
    __tablename__ = "asset_software_catalog"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    asset_type: Mapped[str | None] = mapped_column(String(64))
    owner: Mapped[str | None] = mapped_column(String(64))
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    it_contact: Mapped[str | None] = mapped_column(String(64))
    it_contact_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    status: Mapped[str] = mapped_column(String(32), nullable=False, default="active")
    remark: Mapped[str | None] = mapped_column(Text)
    # 「详细介绍」(migration 328) — long-form write-up that doesn't belong in
    # ``remark`` (备注, a one-liner): what the software is for, onboarding
    # notes, internal policy, links. MEDIUMTEXT (16MB) so admins can write
    # as much as they need; Pydantic caps the API surface at 20k chars.
    description: Mapped[str | None] = mapped_column(MEDIUMTEXT)
    # Free-text product version / edition (migration 228). Captured per
    # PCI DSS 「软件资产清单」 — e.g. "Google Workspace Enterprise Standard",
    # "Cloud Service", "2.222.4". Optional, no schema validation: this is
    # the contract / license tier as the admin would write it, not a
    # semver string.
    version: Mapped[str | None] = mapped_column(String(64))
    department: Mapped[str | None] = mapped_column(String(128))
    login_url: Mapped[str | None] = mapped_column(String(512))
    expiry_date: Mapped[date_type | None] = mapped_column(Date)
    license_info: Mapped[str | None] = mapped_column(Text)
    attachment: Mapped[str | None] = mapped_column(String(256))
    custom_fields: Mapped[dict | None] = mapped_column(JSON)
    # Provisioning extensions (migration 021)
    sso_provider: Mapped[str | None] = mapped_column(String(32))
    offboarding_grace_days: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    # Offboarding policy (migration 033) — decides what the platform does
    # with this software's seats when a holder leaves.
    #   ``sso_cascade`` — seats die when Google/Feishu SSO is suspended; no
    #                     action. No connector required. (Default.)
    #   ``auto_revoke`` — call the software's connector driver to revoke.
    #                     Requires a configured ``software_connectors`` row.
    offboarding_policy: Mapped[str] = mapped_column(
        String(32), nullable=False, default="sso_cascade",
    )
    offboarding_config: Mapped[dict | None] = mapped_column(JSON)
    # Self-cancel policy (migration 361) — may a seat HOLDER drop their
    # own seat from the self-service portal / MCP, or is this software
    # IT-managed?
    #   ``True``  (default) — the employee may schedule "cancel next
    #             month" on their own seat. The normal self-service path.
    #   ``False`` — managed software (Google Workspace, Feishu, and other
    #             identity-tier tools): the employee must NOT be able to
    #             self-cancel, because the account is provisioned/managed
    #             by IT and tied to SSO/identity. Asset/global admins can
    #             still cancel on the employee's behalf (admin-on-behalf);
    #             the gate only blocks the self path.
    allow_self_cancel: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default=text("1"),
    )
    # Free-text category that replaces the legacy ``is_ai_tool`` flag.
    # Recognised values today: ``ai`` (chargeback dashboard's source of
    # truth), ``im``, ``mail``, ``cloud``, ``dev``, ``office``,
    # ``security``, ``other`` (fallback). Migration 082 backfills
    # category=ai for every row that was previously is_ai_tool=true and
    # then drops the boolean. New categories are admin-editable from
    # /admin/assets/software-catalog and can be invented freely — the
    # AI-spending dedup only filters on ``category='ai'`` so unknown
    # values are harmless. Default ``other`` keeps the column NOT NULL
    # without forcing an explicit value on every catalog row.
    category: Mapped[str] = mapped_column(
        String(32), nullable=False,
        default="other", server_default=text("'other'"),
    )
    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 AssetSupplier(Base):
    """Supplier master data used by procurement and software payments."""
    __tablename__ = "asset_suppliers"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    name: Mapped[str] = mapped_column(String(128), nullable=False, unique=True)
    contact_name: Mapped[str | None] = mapped_column(String(64))
    contact_phone: Mapped[str | None] = mapped_column(String(32))
    remark: Mapped[str | None] = mapped_column(Text)
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=text("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 AssetBusinessPayment(Base):
    """Business payment tracking for software subscriptions."""
    __tablename__ = "asset_business_payments"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    software_catalog_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("asset_software_catalog.id"),
    )
    procurement_order_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("asset_procurement_orders.id"),
    )
    supplier_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("asset_suppliers.id"),
    )
    software_name: Mapped[str] = mapped_column(String(128), nullable=False)
    software_type: Mapped[str | None] = mapped_column(String(64))
    payment_frequency: Mapped[str | None] = mapped_column(String(64))
    payment_method: Mapped[str | None] = mapped_column(String(64))
    signing_entity: Mapped[str | None] = mapped_column(String(128))
    po_number: Mapped[str | None] = mapped_column(String(64))
    contract_expiry: Mapped[str | None] = mapped_column(String(128))
    requester: Mapped[str | None] = mapped_column(String(64))
    requester_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    amount: Mapped[Decimal | None] = mapped_column(Numeric(12, 2))
    remark: 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 AssetLegalEntity(Base):
    """Company legal entity information (tax, bank, etc)."""
    __tablename__ = "asset_legal_entities"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    name: Mapped[str] = mapped_column(String(128), nullable=False, unique=True)
    # Short tag-safe identifier (CN / US / SG / JP) embedded in the
    # asset-tag auto-generation rule:
    # ``IT-{purchase_entity.code}-{category code}-{YYYYMMDDHHMMSS}``.
    # Nullable so existing rows backfill as NULL — intake refuses to
    # auto-generate a tag against an entity with no code (422), which
    # nudges admins to fill it via the OfficeAddress 公司主体信息 tab.
    # NOT unique on purpose (migration 133 dropped the unique index):
    # multiple Chinese subsidiaries legitimately share ``CN`` as the
    # tag-prefix slug; tag uniqueness is enforced at the asset row
    # level by ``asset_items.asset_tag`` UNIQUE.
    code: Mapped[str | None] = mapped_column(String(16))
    country_or_region: Mapped[str | None] = mapped_column(String(64))
    tax_identifier: Mapped[str | None] = mapped_column(String(128))
    registered_address: Mapped[str | None] = mapped_column(String(512))
    phone: Mapped[str | None] = mapped_column(String(64))
    bank_name: Mapped[str | None] = mapped_column(String(128))
    bank_account: Mapped[str | None] = mapped_column(String(128))
    bank_account_name: Mapped[str | None] = mapped_column(String(128))
    bank_address: Mapped[str | None] = mapped_column(String(512))
    swift_code: Mapped[str | None] = mapped_column(String(32))
    currency: Mapped[str | None] = mapped_column(String(8))
    account_type: Mapped[str | None] = mapped_column(String(64))
    notes: Mapped[str | None] = mapped_column(Text)
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=text("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 AssetOfficeAddress(Base):
    """Office addresses for each city.

    The 当地 Admin list lives entirely in the
    ``asset_office_local_admins`` join table, with one row flagged
    ``is_primary``. A denormalised ``admin_person_id`` column mirrored
    that primary until migration 540 — two places recording one fact,
    written by a single code path and read by three others.
    Single-contact callers use
    ``office_local_admins.get_office_primary_local_admin()``.

    As of migration 535 these people carry **no asset authority**:
    the ``asset_admin`` role and every regional write gate derive
    from ``asset_location_admins`` instead. An office's 当地 Admin is
    the local IT contact for the humans who sit there, nothing more.
    """
    __tablename__ = "asset_office_addresses"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    city: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
    office_name: Mapped[str | None] = mapped_column(String(128))
    country_or_region: Mapped[str | None] = mapped_column(String(64))
    contact_name: Mapped[str | None] = mapped_column(String(64))
    contact_phone: Mapped[str | None] = mapped_column(String(32))
    address: Mapped[str | None] = mapped_column(String(512))
    address_en: Mapped[str | None] = mapped_column(String(512))
    legal_entity_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("asset_legal_entities.id"),
    )
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=text("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 AssetOfficeLocalAdmin(Base):
    """Many-to-many: which persons are the 当地 Admin of which office.

    An office's local IT contact. Read only by the person-anchored
    flows — the Day-1 welcome CC audience, the onboarding hardware
    recipient, and the offboarding handover suggestion — all of which
    key off where a *person* sits, not where an *asset* lives.

    This table granted the ``asset_admin`` role until migration 535
    (when it was still named ``asset_office_admins``). It no longer
    does: asset authority derives from ``AssetLocationAdmin``. Nothing
    here scopes an asset write, and editing this list changes no
    role. Renamed at the same time so the two tables can't be
    mistaken for one another.

    Composite PK on (office_id, person_id). ``is_primary`` marks the
    row that single-contact callers address.
    """
    __tablename__ = "asset_office_local_admins"
    __mapper_args__ = {"eager_defaults": True}

    office_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("asset_office_addresses.id", ondelete="CASCADE"),
        primary_key=True,
    )
    person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id", ondelete="CASCADE"),
        primary_key=True,
    )
    is_primary: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default=text("0"),
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )


class EmployeeRequest(Base):
    """Employee self-service requests — unified table for all request types.

    Originally ``asset_employee_requests`` (assets only); renamed in
    migration 030 to be the single home for self-service requests across
    domains (laptop replacement, computer borrow, asset return,
    offboarding buyback, software seat, api_key_request, and anything
    added later). Domain-specific fields live in ``detail`` (JSON) and,
    for the api_key case, in the linked ``api_key_assignments`` row via
    ``api_key_assignments.employee_request_id``.
    """
    __tablename__ = "employee_requests"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )
    request_number: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    request_type: Mapped[str] = mapped_column(String(32), nullable=False)
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending")

    requester_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
        nullable=False,
    )
    asset_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("asset_items.id"),
    )
    category: Mapped[str | None] = mapped_column(String(32))
    target_user_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    reason: Mapped[str | None] = mapped_column(Text)
    detail: Mapped[Any | None] = mapped_column(JSON)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
    )
    processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    processed_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id"),
    )
    notes: Mapped[str | None] = mapped_column(Text)


# Backward-compat alias — in-progress callers still reference the old name.
# Delete once ApiKeyService dual-write lands and the name has propagated.
AssetEmployeeRequest = EmployeeRequest


class SoftwareLicensePlan(Base):
    """License plan (seat type) for a software catalog entry.

    Each software can have multiple plans, e.g. "Basic ¥100/mo", "Pro ¥200/yr".
    """
    __tablename__ = "software_license_plans"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # ondelete defaults to RESTRICT — deleting a software_catalog row
    # while it still has plans is refused at both the application
    # layer (delete_software_catalog refuses with 409 when plans exist)
    # and at the DB layer. The previous shape used CASCADE, which silently
    # vaporised every plan when admin clicked "delete software" — and
    # because ``software_monthly_bills.plan_id`` is a bare UUID with no
    # FK (audit-trail isolation), historical bills were left dangling.
    # See migration 119 for the FK rename.
    software_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("asset_software_catalog.id"),
        nullable=False, index=True,
    )
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    price: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    billing_cycle: Mapped[str] = mapped_column(
        String(16), nullable=False,
    )  # "monthly", "yearly", "one_time"
    description: Mapped[str | None] = mapped_column(Text)
    # Seat quota. NULL = unlimited (default). Integer = hard cap enforced
    # by the provisioning service on grant; requests exceeding the cap
    # are routed to an "over-quota approval" path instead of dispatched
    # to the driver (see provisioning service `enforce_plan_quota`).
    max_seats: Mapped[int | None] = mapped_column(Integer, nullable=True)
    # Policy when a grant would push the plan over ``max_seats``:
    #   ``reject``    — default; raise QUOTA_EXCEEDED at request time
    #   ``evict_lru`` — accept the request, approval proceeds; at grant
    #                   dispatch the worker evicts the least-recently-
    #                   logged-in seat holder (with a whitelist) and
    #                   then runs the grant. Used for Zoom.
    over_quota_policy: Mapped[str] = mapped_column(
        String(16), nullable=False, default="reject", server_default="reject",
    )
    # ISO 4217 currency code for ``price``. Defaults to ``CNY`` to
    # preserve historical billing aggregation behaviour for plans
    # created before the column existed. Mixed-currency aggregation
    # is refused at the service layer (see provisioning.compute_billing
    # and asset.billing_service); admin must split a multi-currency
    # tenant into per-currency totals.
    currency: Mapped[str] = mapped_column(
        String(3), nullable=False, default="CNY", server_default="CNY",
    )
    status: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
    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(),
    )

    software: Mapped["AssetSoftwareCatalog"] = relationship(lazy="joined")


class SoftwareSeatAllocation(Base):
    """Records a seat allocated to a user for a specific license plan.

    Created after approval; tracks cost attribution per person/department.
    """
    __tablename__ = "software_seat_allocations"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    plan_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("software_license_plans.id"), nullable=False, index=True,
    )
    user_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False, index=True,
    )
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="active",
    )  # "active", "revoked"
    reason: Mapped[str | None] = mapped_column(Text)
    approval_request_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    allocated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Self-service "cancel at end of next billable period" marker.
    # Set when an employee (or admin on their behalf) requests the seat
    # be revoked at a future month boundary instead of immediately —
    # they keep the seat for the rest of the period they've already
    # billed for, then the ``software_seat.auto_revoke_pending_cancels``
    # cron job picks them up and runs the standard revoke path. NULL
    # when no cancellation is pending. The companion ``cancel_requested``
    # fields capture who/when so the audit trail survives even after
    # the cron actually revokes (it logs ``actor_user_id`` as the
    # original requester, not the system actor).
    cancel_at_period_end: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    cancel_requested_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    cancel_requested_by: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    cancel_reason: Mapped[str | None] = mapped_column(Text)
    # Drift cleanup grace tracking. Stamped by a seat-sync handler
    # the first time a seat's vendor-side membership is found
    # missing; cleared again the moment the membership reappears.
    # Handlers that revoke on drift compare ``now - drift_observed_at``
    # against the per-vendor grace window. NULL means "currently in
    # roster (or never observed missing)". Used by feishu.seat_sync;
    # other vendors may opt in later.
    drift_observed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # ── Idle-seat reclaim (software_seat.idle_reclaim) ──────────────
    # Lifecycle: the daily idle job stamps ``idle_warned_at`` the first
    # time a seat crosses the idle threshold (start of the grace
    # buffer); after the buffer elapses with the seat still idle it is
    # reclaimed (status→revoked). ``idle_kept_until`` is the employee's
    # (or admin's) self-service "keep my seat" opt-out — while it's set
    # and in the future the job skips the seat entirely. Keeping clears
    # ``idle_warned_at`` so that when the keep window later expires the
    # employee gets a fresh warning + full buffer rather than an instant
    # reclaim. ``idle_manual_review_at`` records when the job punted a
    # seat to a human because removing the bound group alone would NOT
    # fully revoke the product (the holder has it via another group too)
    # — it dedups the admin notification so the same seat isn't re-paged
    # every tick.
    idle_warned_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    idle_kept_until: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    idle_manual_review_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(),
    )

    plan: Mapped["SoftwareLicensePlan"] = relationship(lazy="joined")


class SoftwareExternalHolder(Base):
    """A vendor license holder that DEVIATES from "active Helix employee".

    Written by every ``*.seat_sync`` handler: after a sync lists the
    real license holders on the vendor side (Atlassian product_access,
    Zoom roster, Google tenant, Feishu roster, Slack members) and
    classifies each into one of 6 mutually-exclusive buckets, the rows
    persisted here are the four that DEVIATE from the in-use baseline
    (``employees_active`` — Helix-active person on the paid site with
    a healthy vendor account — never row-persisted; it's the "everything
    is fine" category). Together with ``software_seat_sync_stat`` they
    let the operator answer "WHO is occupying each billable seat" not
    just "how many".

    **Category discriminator (since migration 219)** — ``category`` is
    one of:

      * ``external`` — vendor account whose email doesn't match any
        Helix person (partners, contractors, service / bot accounts,
        ex-employees deleted from HR). The original 2026-05-28 use
        case; ``person_id`` is NULL.
      * ``leaver_residual`` — vendor account whose email matches a
        Helix person whose ``employment_status`` is ``left`` /
        ``leaver_pending``. The seat should've been revoked by
        offboarding but didn't (SCIM bounce, default-group reassertion,
        manual re-add). Highest operational severity — a leaver still
        has paid access. ``person_id`` points to the leaver.
      * ``vendor_inactive`` — vendor account_status is NOT ``active``
        (Atlassian ``inactive``, Zoom ``deactivated``, Slack ``deleted``,
        etc.) yet the account still occupies a billable seat per the
        vendor's billing UI. Often a transitional state during
        offboarding (vendor side flipped, billing side hasn't released
        yet). ``person_id`` populated when matched.
      * ``off_paid_site`` — Atlassian-specific. Vendor account is
        ``active`` and holds the product, but only on a NON-paid site
        (sandbox / personal / external Atlassian instances).
        For marketplace add-ons (Tempo, ScriptRunner) these still get
        billed org-wide; for native products they don't. Surfaced so
        the operator can audit "who has a sandbox seat we're not
        tracking on the paid tenant".

    A 5th category ``no_email`` (vendor accounts with no email at all —
    service / API accounts, hand-created consultant accounts) is counted
    in ``SoftwareSeatSyncStat.holders_no_email`` and, since 2026-07-02,
    MAY additionally be row-persisted when the handler can supply a
    vendor-side ``holder_key`` (Feishu: ``feishu-user:<user_id>``;
    ``email`` is NULL on those rows). Handlers without a stable vendor
    key remain count-only. Persisting them makes the account visible +
    annotatable in the external-accounts directory instead of being an
    anonymous counter.

    Lifecycle: upserted (``last_observed_at`` bumped, ``resolved=False``)
    every run the holder is still in this category. A holder that
    disappears from the vendor side OR changes category (e.g. external
    → leaver_residual when the email is finally HRIS'd in as someone
    who left) is marked ``resolved=True`` for the prior category by
    the same run that no longer observes them THERE; a new row appears
    under the new category. Resolved rows are kept for history.

    The unique key is (plan_id, holder_key). For per-USER vendors
    ``holder_key == email`` (one email is in at most one category per
    plan, category transitions upserted in place — historical behaviour).
    For per-DEVICE vendors (Jamf MDM) ``holder_key`` is the device-unique
    key (serial) so a single employee holding several devices yields
    several rows / seats instead of collapsing on a shared email.
    """
    __tablename__ = "software_external_holder"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        # One row per (plan, holder_key) — the upsert key. holder_key is
        # ``email`` for per-user vendors and the device key for per-device
        # vendors (Jamf). Category transitions upsert in place.
        Index(
            "uq_software_external_holder_plan_holder_key",
            "plan_id", "holder_key", unique=True,
        ),
        # "Show me this software's open external holders" — the drawer
        # + the company-wide reconciliation aggregation both filter on
        # (software_id, resolved).
        Index(
            "ix_software_external_holder_software_resolved",
            "software_id", "resolved",
        ),
        Index(
            "ix_software_external_holder_plan_resolved",
            "plan_id", "resolved",
        ),
        # Drawer per-category tab — pull a single category from a plan.
        Index(
            "ix_software_external_holder_plan_category",
            "plan_id", "category",
        ),
        # Reverse lookup: "which residual seats does this leaver still
        # hold?" — used by the offboarding view's per-person residual
        # badge.
        Index(
            "ix_software_external_holder_person_id",
            "person_id",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("integrations.id"), nullable=False, index=True,
    )
    software_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("asset_software_catalog.id"), nullable=False,
    )
    plan_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("software_license_plans.id"), nullable=False,
    )
    # Vendor short key: 'jira' | 'zoom' | 'google' | 'feishu' | 'slack'.
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    # Product family / SKU on the vendor side (e.g. 'jira-software',
    # 'jira-service-desk', 'confluence', 'zoom-base'). NULL for vendors
    # with a single undifferentiated roster (Slack).
    product_key: Mapped[str | None] = mapped_column(String(64))
    # Vendor site / workspace the holder appears on (Atlassian site,
    # Slack workspace, Feishu group chat id). NULL when not applicable.
    # For category='off_paid_site' this stores the NON-paid site host
    # (sandbox / personal) so the drawer shows "where" the seat lives.
    site: Mapped[str | None] = mapped_column(String(255))
    # Dedup/uniqueness key (since holder_key migration). For per-user
    # vendors this equals ``email``; for per-device vendors (Jamf) it's
    # the device-unique key (serial). Always populated. Backfilled to
    # ``email`` for pre-migration rows. The ORM default falls back to the
    # row's (lowercased) ``email`` when an inserter doesn't set it — so
    # per-user callers/factories that only know ``email`` keep working
    # with the historical (plan, email) uniqueness semantics.
    holder_key: Mapped[str] = mapped_column(
        String(255), nullable=False,
        default=lambda ctx: (
            (ctx.get_current_parameters().get("email") or "").strip().lower()
            or None
        ),
    )
    # The holder's email — for display + person-linking. Stored
    # lowercased. NULLABLE now that ``holder_key`` is the join key (a
    # per-device holder row may carry a null email if the device has no
    # assigned user; ``category='no_email'`` rows are keyed on a vendor
    # id like ``feishu-user:<user_id>``; per-user vendors otherwise set
    # it == holder_key). Holders with no dedupe key at all are NOT
    # row-persisted; counted in
    # ``SoftwareSeatSyncStat.holders_no_email`` instead.
    email: Mapped[str | None] = mapped_column(String(255), nullable=True)
    # Mutually-exclusive category — see class docstring. Default
    # ``'external'`` for back-compat with pre-217 rows; new categories
    # are written explicitly by the handler.
    category: Mapped[str] = mapped_column(
        String(32), nullable=False, default="external",
        server_default=text("'external'"),
    )
    # Helix person FK — NULL for ``category='external'`` (definitionally
    # no match); populated for ``leaver_residual`` / ``vendor_inactive`` /
    # ``off_paid_site`` when the email resolved via the alias index.
    # ON DELETE SET NULL so deleting a person archives the residual
    # observation rather than the row vanishing silently.
    person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
        nullable=True,
    )
    display_name: Mapped[str | None] = mapped_column(String(255))
    # Vendor-side stable id (Atlassian accountId, Zoom user id, Google
    # id, Feishu union_id, Slack user id) — powers the "open in vendor
    # admin console" deep-link where the vendor has a stable URL.
    vendor_account_id: Mapped[str | None] = mapped_column(String(128))
    # Vendor account status ('active', 'suspended', 'inactive', ...) so
    # the drawer can flag e.g. a suspended account still holding a seat.
    account_status: Mapped[str | None] = mapped_column(String(32))
    # Last activity date the vendor reports for this holder on this
    # product, if any. NULL = vendor never reported activity.
    last_active_on: Mapped[date_type | None] = mapped_column(Date)
    first_observed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    last_observed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    resolved: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, server_default=text("0"),
    )
    resolved_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 SoftwareSeatSyncStat(Base):
    """Per-plan reconciliation snapshot from the latest ``*.seat_sync`` run.

    One row per license plan, upserted each sync. Holds the authoritative
    headline numbers — every billable seat the vendor counts is bucketed
    into exactly one of six mutually-exclusive categories, so the operator
    can answer "WHO is occupying each seat" not just "how many":

      * ``vendor_license_holders`` — **total accounts the vendor bills**
        for this product, org-wide (matches Atlassian's Admin UI count
        modulo timing slip). The six bucket columns sum to this.
      * ``holders_matched`` — vendor account is ``active``, email maps
        to a Helix person whose ``employment_status='active'``, and
        the holder is on the plan's paid site. "Employees properly
        using their seat" — the desired baseline; not row-persisted
        in ``SoftwareExternalHolder``.
      * ``holders_external`` — vendor account is ``active``, email
        matches no Helix person (partners, contractors, service /
        bot accounts). Persisted with ``category='external'``.
      * ``holders_no_email`` — vendor account with no email at all
        (service / API accounts). Count-only — not row-persisted
        because there's no dedupe key.
      * ``holders_leaver_residual`` (since migration 219) — vendor
        account is ``active`` and the email matches a Helix person
        whose ``employment_status='left'`` / ``leaver_pending``.
        Offboarding leak — the seat should have been revoked by
        the offboarding flow but wasn't. Highest operational
        severity. Persisted with ``category='leaver_residual'``.
      * ``holders_vendor_inactive`` (since migration 219) — vendor
        ``account_status`` is NOT ``active`` (Atlassian ``inactive``,
        Zoom ``deactivated``, Slack ``deleted``) yet the account
        still occupies a billable seat per the vendor's billing UI.
        Transitional during offboarding. Persisted with
        ``category='vendor_inactive'``.
      * ``holders_off_paid_site`` (since migration 219) — Atlassian-
        specific. Vendor account is ``active`` and holds the product,
        but only on a NON-paid site (sandbox / personal / external
        Atlassian instance). Marketplace addons (Tempo, ScriptRunner)
        still bill these org-wide. Persisted with
        ``category='off_paid_site'``.

      * ``synced_at`` / ``run_id`` — provenance so the UI can say "as of
        <time>" instead of leaving staleness ambiguous.

    **Identity (NON-NEGOTIABLE)**: ``vendor_license_holders ==
    holders_matched + holders_external + holders_no_email +
    holders_leaver_residual + holders_vendor_inactive +
    holders_off_paid_site``. The shared
    :func:`record_seat_reconciliation` helper raises
    :class:`SeatReconciliationIdentityError` at write time if a caller
    violates this — see the helper's docstring for the rationale.

    Occupancy (active seats) and scheduled-cancel counts are queried live
    from ``software_seat_allocations`` — they change between syncs (an
    employee can self-cancel any time) so freezing them here would lie.
    """
    __tablename__ = "software_seat_sync_stat"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index(
            "uq_software_seat_sync_stat_plan",
            "plan_id", unique=True,
        ),
        Index("ix_software_seat_sync_stat_software", "software_id"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    integration_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("integrations.id"), nullable=False, index=True,
    )
    software_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("asset_software_catalog.id"), nullable=False,
    )
    plan_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("software_license_plans.id"), nullable=False,
    )
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    product_key: Mapped[str | None] = mapped_column(String(64))
    vendor_license_holders: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    holders_matched: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    holders_external: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    holders_no_email: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    # ── Migration 217 (2026-05-29): three more buckets so the
    # displayed identity matches Atlassian's billing UI count, not
    # the filtered-down subset Helix used to show ──────────────────
    # Vendor account is `active` and holds the product on a paid site,
    # the email maps to a Helix person whose ``employment_status`` is
    # ``left`` / ``leaver_pending``. Offboarding leak — needs IT action.
    holders_leaver_residual: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    # Vendor account_status is NOT ``active`` (Atlassian ``inactive``,
    # Zoom ``deactivated``, Slack ``deleted``) but still occupies a
    # billable seat per the vendor's billing UI. Often transitional
    # during offboarding (vendor deactivated, billing not yet released).
    holders_vendor_inactive: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    # Vendor account is ``active`` and holds the product, but only on
    # a NON-paid site (sandbox / personal / external Atlassian site).
    # For marketplace addons (Tempo, etc.) these still get billed
    # org-wide; surfaced so the operator can audit them.
    holders_off_paid_site: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    synced_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    # ``scheduled_job_runs.id`` of the run that produced this snapshot —
    # bare UUID (no FK; run rows are pruned on their own retention clock
    # and we don't want a snapshot delete to cascade from that).
    run_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    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 SoftwareMonthlyBill(Base):
    """Frozen monthly bill line for one seat allocation in one period.

    Generated by the admin "close month" action. Once written, these
    rows are immutable (reopen deletes + close re-creates — never
    edits in place). See migration 063 for the design rationale.
    """
    __tablename__ = "software_monthly_bills"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index("ix_software_monthly_bills_period", "period"),
        Index("ix_software_monthly_bills_user_id", "user_id"),
        Index("ix_software_monthly_bills_software_id", "software_id"),
        # Seat-line idempotency (migration 063; previously DB-only —
        # declared here so the model tells the truth).
        UniqueConstraint(
            "period", "allocation_id",
            name="uq_software_monthly_bills_period_alloc",
        ),
        # Addon-line idempotency (migration 382): allocation_id is NULL
        # for addon lines, and MySQL UNIQUE treats every NULL as
        # distinct — so the alloc key gave addon lines ZERO protection
        # against a double close inserting them twice. charge_id is
        # non-NULL exactly for addon lines, so this key is the addon
        # twin of the alloc key (seat lines have charge_id NULL and
        # pass freely).
        UniqueConstraint(
            "period", "charge_id",
            name="uq_software_monthly_bills_period_charge",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # "YYYY-MM" — string form so period equality is exact.
    period: Mapped[str] = mapped_column(String(7), nullable=False)
    # Line discriminator (migration 348). ``seat`` = the historical line
    # type, one row per SoftwareSeatAllocation per period. ``addon`` = a
    # one-time software add-on charge (AI credits, vendor person-days,
    # …) materialised here so every downstream cost rollup
    # (finance_spend_coordinator, software-cost views, my-cost) picks it
    # up via the same ``SUM(amount)`` with zero per-query changes — the
    # money cannot be lost in a parallel table. Defaults to ``seat`` so
    # every pre-348 row reads correctly without a backfill.
    line_type: Mapped[str] = mapped_column(
        String(16), nullable=False, default="seat", server_default=text("'seat'"),
    )
    # allocation_id / plan_id are NULL for ``addon`` lines — an add-on
    # charge has no seat allocation and no license plan. NOT NULL was
    # relaxed in migration 348; seat lines still always carry both.
    allocation_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    software_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    plan_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # Set only on ``addon`` lines — the source SoftwareAddonCharge.id.
    charge_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # Quantity for ``addon`` lines (e.g. 3 person-days, 2× credit packs).
    # NULL for seat lines.
    quantity: Mapped[int | None] = mapped_column(Integer)
    user_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    software_name: Mapped[str | None] = mapped_column(String(128))
    plan_name: Mapped[str | None] = mapped_column(String(128))
    user_name: Mapped[str | None] = mapped_column(String(128))
    department: Mapped[str | None] = mapped_column(String(128))
    reason: Mapped[str | None] = mapped_column(Text)
    price: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    # ``monthly`` / ``yearly`` / ``one_time`` for seat lines; always
    # ``one_time`` for addon lines.
    billing_cycle: Mapped[str] = mapped_column(String(16), nullable=False)
    # What actually counts toward this period's total. Pre-computed
    # at close time so sum queries can just SUM(amount) without
    # branching on billing_cycle — see migration 063.
    amount: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    # ISO 4217 currency code copied from the plan at close-time. Frozen
    # alongside ``amount`` so a later admin edit to ``plans.currency``
    # cannot retroactively change the currency of an already-finalised
    # bill row. Defaults to ``CNY`` to preserve historical aggregation
    # semantics for rows written before the column existed (migration
    # 116 backfills explicitly to CNY). Cross-currency aggregation is
    # rejected at the service layer — see ``billing_service`` and
    # ``provisioning.compute_billing``.
    currency: Mapped[str] = mapped_column(
        String(3), nullable=False, default="CNY", server_default=text("'CNY'"),
    )
    allocated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    finalized_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    finalized_by: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )


class AssetInventoryCycle(Base):
    """A 资产盘点 campaign — one row per round.

    Snapshots a set of (recipient, asset) pairs at creation time into
    ``asset_inventory_items``. Subsequent asset changes do NOT alter
    the snapshot — items added later (e.g. a new claim today) are
    captured by the next cycle, not retro-injected into this one.

    See ``DECISIONS.md`` (2026-04-29) for the routing model
    (assignee / location_admin / it_fallback) and the response-option
    matrix.
    """

    __tablename__ = "asset_inventory_cycles"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index("ix_asset_inv_cycles_status", "status"),
        Index("ix_asset_inv_cycles_started_at", "started_at"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    # Free-text description shown to recipients in the Feishu card.
    description: Mapped[str | None] = mapped_column(Text)
    # ``draft`` (created but not started) → ``active`` (snapshotted +
    # notifications dispatched) → ``closed`` (admin closed it; remaining
    # items get status='pending' frozen and last_inventory_at on assets
    # is left untouched for those).
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="draft", server_default="draft",
    )
    # JSON: {"departments": [...], "locations": [...], "categories":
    # [...], "person_ids": [...]}. Empty / missing keys mean "no filter
    # on that dimension". Empty object = company-wide.
    scope_filter: Mapped[Any | None] = mapped_column(JSON, default=dict)
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Whether the auto-reminder T+7 push has already gone out for this
    # cycle. Single-shot flag — v1 does NOT do repeated reminders.
    reminder_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # NULL = system-triggered (scheduled job). Non-NULL = admin-triggered.
    created_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    closed_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    # Cached counters — recomputed by service.recompute_counters
    # whenever items mutate. Avoid count(*) on every list render.
    total_items: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    pending_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    confirmed_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )
    issue_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 AssetInventoryItem(Base):
    """One (cycle × asset) snapshot row.

    ``holder_kind``:
    * ``assignee`` — asset has a holder; ``holder_person_id`` = holder
    * ``location_admin`` — idle asset routed to the administrator of
      its managed location (``office_admin`` before migration 543,
      when the recipient was still the office's 当地 Admin);
      ``holder_person_id`` = office admin, ``office_id`` = matched row
    * ``it_fallback`` — idle asset with no resolvable office admin OR
      asset whose assignee is inactive; both holder fields NULL. IT
      handles these from a dedicated admin tab.

    Asset/holder fields are denormalised at snapshot time so the cycle
    detail view doesn't need a 6-table join, and so a later
    rename/transfer of the asset doesn't retroactively change what the
    recipient was actually asked to confirm.
    """

    __tablename__ = "asset_inventory_items"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        # An asset appears at most once in a given cycle — see the
        # snapshot fan-out in ``inventory_service.create_cycle``.
        Index(
            "uq_asset_inv_items_cycle_asset",
            "cycle_id", "asset_id",
            unique=True,
        ),
        Index("ix_asset_inv_items_cycle_status", "cycle_id", "status"),
        Index("ix_asset_inv_items_holder", "holder_person_id"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    cycle_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("asset_inventory_cycles.id", ondelete="CASCADE"),
        nullable=False,
    )
    asset_id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        ForeignKey("asset_items.id", ondelete="CASCADE"),
        nullable=False,
    )
    # ``assignee`` | ``location_admin`` | ``it_fallback``
    # (pre-543 rows may still read ``office_admin`` — see
    # ``HOLDER_KINDS_LOCATION_ADMIN``)
    holder_kind: Mapped[str] = mapped_column(String(16), nullable=False)
    holder_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    office_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("asset_office_addresses.id"),
    )

    # ── Snapshot of asset shape at cycle-creation time ──
    asset_tag_snapshot: Mapped[str | None] = mapped_column(String(64))
    asset_name_snapshot: Mapped[str | None] = mapped_column(String(128))
    asset_category_snapshot: Mapped[str | None] = mapped_column(String(32))
    asset_location_snapshot: Mapped[str | None] = mapped_column(String(256))
    asset_status_snapshot: Mapped[str | None] = mapped_column(String(32))

    # ``pending`` | ``confirmed`` | ``issue_reported``
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="pending", server_default="pending",
    )
    # Allowed values depend on ``holder_kind`` — see service.respond.
    response_type: Mapped[str | None] = mapped_column(String(32))
    response_notes: Mapped[str | None] = mapped_column(Text)
    # When ``response_type='transferred'`` (assignee bucket).
    transferred_to_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    # When ``response_type='moved_to'`` (location_admin bucket).
    new_location: Mapped[str | None] = mapped_column(String(256))

    responded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    responded_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    reminded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # Issue-resolution tracking (only meaningful when
    # ``status='issue_reported'``). ``issue_status`` is ``open`` |
    # ``resolved``; admin resolves manually after taking action
    # (raise approval, retag, write off, …).
    issue_status: Mapped[str | None] = mapped_column(String(16))
    issue_resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    issue_resolved_by: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"),
    )
    issue_resolution_notes: Mapped[str | None] = mapped_column(Text)

    # S3 object keys for any evidence photos the holder uploaded with
    # their response — only meaningful when ``response_type='not_mine'``
    # ("记录有误，不属于我"), where we require proof of the actual hardware
    # SN. Stored as a JSON array of strings; resolve to presigned GET
    # URLs at render time via ``app.core.storage.presign_get``.
    evidence_photos: Mapped[list[str] | None] = mapped_column(JSON)

    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 SoftwareMonthlyBillHistory(Base):
    """Snapshot of finalised ``software_monthly_bills`` rows captured by
    ``billing_service.reopen_month`` BEFORE the period is cleared and
    re-closed. Append-only: rows are inserted, never edited.

    Lets the audit trail answer "what did we previously bill for
    2026-04, and who reopened / re-closed it?". One reopen produces one
    ``supersedes_seq`` generation per period (first reopen = 1, second
    reopen = 2, …); the surviving live bill row in
    ``software_monthly_bills`` is always the most recent re-close.

    Schema mirrors ``SoftwareMonthlyBill`` plus the supersession
    metadata. See migration 116 for column rationale and the
    ``cmdb_billing_alerts_dedup`` companion table.
    """
    __tablename__ = "software_monthly_bill_history"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index(
            "ix_software_monthly_bill_history_period_seq",
            "period", "supersedes_seq",
        ),
        Index("ix_software_monthly_bill_history_period", "period"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    period: Mapped[str] = mapped_column(String(7), nullable=False)
    # Generation counter for this period — first reopen=1, second=2, …
    # Together with ``period`` uniquely identifies a snapshot batch.
    supersedes_seq: Mapped[int] = mapped_column(Integer, nullable=False)
    # Original bill row id from ``software_monthly_bills`` at snapshot
    # time. Kept as a plain Uuid (no FK) because the source row is
    # deleted in the same transaction.
    original_bill_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    # Mirrors SoftwareMonthlyBill (migration 348): seat lines carry
    # allocation_id + plan_id; addon lines carry charge_id + quantity and
    # leave allocation_id / plan_id NULL.
    line_type: Mapped[str] = mapped_column(
        String(16), nullable=False, default="seat", server_default=text("'seat'"),
    )
    allocation_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    software_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    plan_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    charge_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    quantity: Mapped[int | None] = mapped_column(Integer)
    user_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    software_name: Mapped[str | None] = mapped_column(String(128))
    plan_name: Mapped[str | None] = mapped_column(String(128))
    user_name: Mapped[str | None] = mapped_column(String(128))
    department: Mapped[str | None] = mapped_column(String(128))
    reason: Mapped[str | None] = mapped_column(Text)
    price: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    billing_cycle: Mapped[str] = mapped_column(String(16), nullable=False)
    currency: Mapped[str] = mapped_column(
        String(3), nullable=False, default="CNY", server_default=text("'CNY'"),
    )
    amount: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    allocated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # When the *original* bill row was finalised (carried forward from
    # ``SoftwareMonthlyBill.finalized_at``) and by whom. Lets finance
    # reconstruct who originally signed off on the now-superseded bill.
    original_finalized_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    original_finalized_by: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # When the snapshot was taken (i.e. when reopen_month ran) and who
    # invoked the reopen. ``superseded_by`` is the actor that triggered
    # the reopen, mirroring ``finalized_by`` semantics on the live row.
    superseded_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    superseded_by: Mapped[uuid.UUID | None] = mapped_column(Uuid)


class SoftwareAddonItem(Base):
    """A configurable one-time add-on item on a software catalog entry.

    Sits alongside ``software_license_plans`` but models *consumption*,
    not a held seat: AI credit packs, vendor person-days, one-off
    onboarding fees, etc. The admin configures a **unit** (``unit_label``
    + ``unit_price`` + ``currency``) and a quantity envelope
    (``min_quantity`` / ``step`` / ``max_quantity``); the employee picks
    a quantity and the charge amount is ``unit_price × quantity`` frozen
    at request time.

    Why a separate concept from a ``one_time`` license plan: a plan grant
    creates a ``SoftwareSeatAllocation`` that enters seat reconciliation
    (生效坐席 == vendor license holders), idle-reclaim and offboarding
    revoke. A consumed add-on has no vendor seat to reconcile and nothing
    to revoke on offboarding — modelling it as a seat would pollute all
    three. See migration 348 and DECISIONS.md (2026-06-27).
    """
    __tablename__ = "software_addon_items"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index("ix_software_addon_items_software_id", "software_id"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    # Index declared once in __table_args__ above (no column-level
    # index=True — that would create a duplicate same-named index).
    software_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("asset_software_catalog.id"), nullable=False,
    )
    # Display name, e.g. "AI Credits", "实施人天".
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    # What one unit *is*, e.g. "1000 credits", "人天". Shown next to the
    # quantity picker so the employee understands what they are buying.
    unit_label: Mapped[str] = mapped_column(String(64), nullable=False)
    # Price of ONE unit in ``currency``. Amount billed = unit_price × qty.
    unit_price: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    currency: Mapped[str] = mapped_column(
        String(3), nullable=False, default="CNY", server_default=text("'CNY'"),
    )
    # Quantity envelope. ``min_quantity`` is "配置一个最小的量"; ``step``
    # constrains increments (e.g. credits sold in packs of 1); a NULL
    # ``max_quantity`` means no hard cap (governance still flows through
    # approval). All enforced at request time in addon_service.
    min_quantity: Mapped[int] = mapped_column(
        Integer, nullable=False, default=1, server_default=text("1"),
    )
    step: Mapped[int] = mapped_column(
        Integer, nullable=False, default=1, server_default=text("1"),
    )
    max_quantity: Mapped[int | None] = mapped_column(Integer)
    description: Mapped[str | None] = mapped_column(Text)
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="active", server_default=text("'active'"),
    )
    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(),
    )

    software: Mapped["AssetSoftwareCatalog"] = relationship(lazy="joined")


class SoftwareAddonCharge(Base):
    """An employee's one-time add-on purchase request and its lifecycle.

    Parallels ``SoftwareSeatAllocation`` but is NOT a seat: it never
    enters seat reconciliation, idle-reclaim or offboarding revoke. Money
    is "spent once" — on approval the charge is pinned to a billing
    ``period`` and ``close_month`` materialises it into one
    ``software_monthly_bills`` row (``line_type='addon'``).

    Status machine: ``pending`` → ``approved`` | ``rejected`` |
    ``cancelled``. Unlike seats there is NO per-(user, item) dedup guard —
    repeat purchases (buy more credits) are legitimate.
    """
    __tablename__ = "software_addon_charges"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        Index("ix_software_addon_charges_user_id", "user_id"),
        Index("ix_software_addon_charges_software_id", "software_id"),
        Index("ix_software_addon_charges_status_period", "status", "period"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    item_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("software_addon_items.id"), nullable=False, index=True,
    )
    # Denormalised from the item at request time so cost rollups and the
    # bill materialisation never need to rejoin items → catalog.
    software_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("asset_software_catalog.id"), nullable=False,
    )
    # user_id index lives in __table_args__ (avoid duplicate same-named idx).
    user_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False,
    )
    quantity: Mapped[int] = mapped_column(Integer, nullable=False)
    # Snapshots frozen at request time — a later admin edit to the item's
    # unit_price / currency must not retroactively change a submitted or
    # billed charge. ``amount`` = unit_price × quantity, pre-computed so
    # the bill materialisation is a straight copy.
    unit_price: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    currency: Mapped[str] = mapped_column(
        String(3), nullable=False, default="CNY", server_default=text("'CNY'"),
    )
    amount: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default="pending", server_default=text("'pending'"),
    )
    reason: Mapped[str | None] = mapped_column(Text)
    approval_request_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # Requester's department snapshot at approval time — the attribution
    # key copied onto the bill row's ``department`` so the BU / finance
    # rollup credits the team that incurred the cost.
    department: Mapped[str | None] = mapped_column(String(128))
    # Billing period (YYYY-MM) the charge is attributed to. Set at
    # approval to the current open period; ``close_month`` materialises
    # approved charges whose ``period`` equals the closing period.
    period: Mapped[str | None] = mapped_column(String(7), index=True)
    approved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    rejected_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    cancelled_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(),
    )

    item: Mapped["SoftwareAddonItem"] = relationship(lazy="joined")


class AssetDepreciationPolicy(Base):
    """Per-category depreciation policy — the rule that turns an asset's
    purchase cost + age into a net book value.

    Keyed on ``asset_category`` (the stable enum stored on
    ``AssetItem.category`` / ``AssetCategoryConfig.asset_category``).
    A single row with ``asset_category IS NULL`` is the **platform
    default fallback** applied to any category without its own policy —
    mirroring the "NULL = wildcard" provisioning-policy convention. The
    service layer guarantees at most one NULL row (MySQL unique indexes
    permit multiple NULLs, so the single-default invariant is enforced
    in code, not the DB).

    No FK on ``asset_category`` — it is an enum string, not a relational
    id (and keeping it FK-free sidesteps TiDB CHAR↔VARCHAR FK strictness).
    """

    __tablename__ = "asset_depreciation_policies"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    # NULL = platform default fallback; otherwise the asset_category enum.
    asset_category: Mapped[str | None] = mapped_column(String(32), unique=True)
    method: Mapped[str] = mapped_column(
        String(32), nullable=False, default="straight_line",
        server_default="straight_line",
    )
    useful_life_months: Mapped[int] = mapped_column(
        Integer, nullable=False, default=36, server_default=text("36"),
    )
    # Salvage value as a 0..1 fraction of cost (e.g. 0.0500 = 5%).
    residual_rate: Mapped[Decimal] = mapped_column(
        Numeric(5, 4), nullable=False, default=Decimal("0.05"),
        server_default=text("0.0500"),
    )
    # Annual rate (0..1) for declining_balance; NULL for straight_line.
    declining_rate: Mapped[Decimal | None] = mapped_column(Numeric(6, 4))
    is_active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default=text("1"),
    )
    notes: 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 AssetBookValueSnapshot(Base):
    """Monthly materialised fleet net-book-value, per category.

    The monthly cron (``asset.book_value_snapshot``) writes one row per
    ``(period, asset_category)`` capturing the active fleet's original
    cost, accumulated depreciation, and net book value AS OF the close of
    that month — all normalised to CNY (via the platform FX rate) so the
    leadership depreciation-trend chart is a single comparable line.

    Materialising the trend (rather than recomputing it live every
    request) survives later asset soft-deletes/disposals: the historical
    fleet value at month-end M is frozen even if the assets are cleaned
    up afterwards. Current/live net book value is still computed on the
    fly (no staleness) — this table backs only the historical trend.
    Idempotent: upsert on the unique ``(period, asset_category)``.
    """

    __tablename__ = "asset_book_value_snapshots"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        # Mirror migration 283's unique key so create_all-based test
        # fixtures build the same table as prod (the upsert in
        # ``snapshot_book_values`` is SELECT-then-INSERT/UPDATE, but the
        # constraint is the final backstop against a concurrent
        # double-insert and keeps model⟷DB schema parity).
        UniqueConstraint(
            "period", "asset_category",
            name="uq_asset_bv_snapshots_period_category",
        ),
        Index(
            "ix_asset_bv_snapshots_period", "period",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    # YYYY-MM — the month whose close this snapshot represents.
    period: Mapped[str] = mapped_column(String(7), nullable=False)
    # The asset_category enum; the per-period total is summed at read.
    asset_category: Mapped[str | None] = mapped_column(String(32))
    original_cost_cny: Mapped[Decimal] = mapped_column(
        Numeric(16, 2), nullable=False, default=Decimal("0"),
        server_default=text("0"),
    )
    accumulated_depreciation_cny: Mapped[Decimal] = mapped_column(
        Numeric(16, 2), nullable=False, default=Decimal("0"),
        server_default=text("0"),
    )
    net_book_value_cny: Mapped[Decimal] = mapped_column(
        Numeric(16, 2), nullable=False, default=Decimal("0"),
        server_default=text("0"),
    )
    asset_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    # How many assets in this bucket carried cost but no purchase_date
    # (counted at full cost, not depreciated) — surfaced for honesty.
    undepreciable_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("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(),
    )


# Classifications that mean "this external account is KNOWN / expected"
# — they are subtracted from the seat-reconciliation "external unmatched"
# anomaly count (still visible when the operator expands the list).
EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS: tuple[str, ...] = (
    "shared_service",    # 公共服务(团队使用) — bot / CI / service account / shared mailbox
    "external_guest",    # CBG保留 — legitimate partner / vendor collaborator
    "contractor",        # 外包 — active contractor not in HRIS
)
# Classifications that still NEED ACTION (kept in the anomaly count).
EXTERNAL_ACCOUNT_ACTION_CLASSIFICATIONS: tuple[str, ...] = (
    "former_employee",   # ghost account — left the company, never offboarded
    "unknown",           # default until an admin triages it
)
EXTERNAL_ACCOUNT_CLASSIFICATIONS: tuple[str, ...] = (
    EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS
    + EXTERNAL_ACCOUNT_ACTION_CLASSIFICATIONS
)


class ExternalAccountRegistry(Base):
    """Cross-vendor annotation for a non-employee account, keyed by email.

    The seat-reconciliation panels of EVERY vendor (Google / Feishu /
    Slack / Jira / Zoom / …) flag accounts that don't match an active
    Helix employee as ``software_external_holder.category='external'``.
    The same email routinely shows up across several vendors at once
    (a service account, a partner, a never-offboarded ex-employee), so
    annotating it per-vendor is redundant. This table is the ONE place
    an admin classifies + notes an email; all vendors' reconciliation
    read it at display time.

    Generalises the older Google-only ``google_orphan_accounts`` "owner
    + purpose" pattern into a vendor-agnostic registry. ``google_orphan_
    accounts`` stays the source of truth for Google account *presence*
    and the offboarding ``RES_GOOGLE_ORPHAN_OWNER`` blocker (owner lives
    there, untouched); this table adds the cross-vendor classification +
    note that every panel consults.

    Key: ``email`` (lowercase, unique). Presence ("which vendors does
    this email appear in") is NOT stored here — it is derived at read
    time from ``software_external_holder`` + ``google_orphan_accounts``.

    **No-email accounts** (since 2026-07-02): vendor accounts with no
    email at all are keyed by their ``software_external_holder.
    holder_key`` (e.g. ``feishu-user:<user_id>``) — that synthetic key
    is stored in ``email`` here. It's still one annotation per stable
    identity; the column name is historical.
    """

    __tablename__ = "external_account_registry"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        UniqueConstraint(
            "email", name="uq_external_account_registry_email",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    # Stored lowercase-normalised; matched case-insensitively against
    # software_external_holder.email / google_orphan_accounts.primary_email.
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    # One of EXTERNAL_ACCOUNT_CLASSIFICATIONS. Drives whether the email
    # is subtracted from each panel's "external unmatched" anomaly count.
    classification: Mapped[str] = mapped_column(
        String(32), nullable=False, default="unknown",
        server_default=text("'unknown'"),
    )
    # Free-text "one annotation" the operator wanted — why this account
    # exists / what to do about it. Shared across all vendors.
    note: Mapped[str | None] = mapped_column(Text)
    # Optional accountable person for NON-Google emails. Google emails
    # keep their owner on google_orphan_accounts.owner_user_id (preserves
    # the offboarding blocker); this column is for the other vendors.
    owner_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid,
        ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )
    created_by: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    updated_by: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    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(),
    )

    def __repr__(self) -> str:
        return (
            f"<ExternalAccountRegistry {self.email} "
            f"[{self.classification}]>"
        )


# ──────────────────────────────────────────────────────────────────
# MDM (Jamf) managed-device ledger
# ──────────────────────────────────────────────────────────────────
# An MDM enrolment is a fact about a DEVICE, not about a person. Jamf
# manages the Mac; the Mac outlives the employee who happened to carry
# it. Helix's Jamf integration originally modelled enrolment as a
# per-person software seat, which made every departure look like an
# access-revocation todo — a todo with nothing to do, because a returned
# laptop is re-imaged and handed to the next hire while staying enrolled
# the whole time. The real cleanup trigger is the device leaving company
# ownership (buyback / sale / write-off), which is an asset-disposal
# event. This table is where that correction lives: one row per managed
# device, keyed by the MDM's own identifiers, with the person only ever
# a derived attribute of the device.
#
# See DECISIONS.md 2026-08-26 (MDM 关联从人改为设备).

# ``match_category`` — how this device's ownership resolved on the last
# sync. Mirrors the shared seat-reconciliation buckets so the same
# classification pass can feed both this ledger and the reconciliation
# snapshot without a second traversal.
MDM_MATCH_MATCHED = "matched"                  # active employee holds it
MDM_MATCH_LEAVER_RESIDUAL = "leaver_residual"  # departed employee still on it
MDM_MATCH_EXTERNAL = "external"                # a non-employee email
MDM_MATCH_UNOWNED = "unowned"                  # pool / lab / shared device
MDM_MATCH_CATEGORIES: frozenset[str] = frozenset({
    MDM_MATCH_MATCHED,
    MDM_MATCH_LEAVER_RESIDUAL,
    MDM_MATCH_EXTERNAL,
    MDM_MATCH_UNOWNED,
})

# ``enrollment_state`` — the device's lifecycle inside the MDM, as Helix
# observes it. NOT a mirror of Jamf's own "managed" boolean (that lives
# in ``managed``); this tracks whether Helix still expects the device to
# be enrolled at all.
MDM_STATE_ENROLLED = "enrolled"
# The asset left company ownership (disposed) while still enrolled — a
# cleanup todo is open against it.
MDM_STATE_PENDING_CLEANUP = "pending_cleanup"
# The device is no longer returned by the MDM inventory. This is the
# GROUND TRUTH for "the admin removed it" — a cleanup that ends here is
# confirmed by the vendor, not by someone clicking a button.
MDM_STATE_GONE = "gone"
# Ordered by how an operator reads the page — what is live, what needs
# doing, what is finished — not alphabetically. The API serves this tuple
# directly and the frozenset below is derived from it, so the display
# order and the membership check cannot drift apart.
MDM_ENROLLMENT_STATE_ORDER: tuple[str, ...] = (
    MDM_STATE_ENROLLED,
    MDM_STATE_PENDING_CLEANUP,
    MDM_STATE_GONE,
)
MDM_ENROLLMENT_STATES: frozenset[str] = frozenset(MDM_ENROLLMENT_STATE_ORDER)


def mdm_serial_key(serial: str | None) -> str | None:
    """Normalise a device serial for cross-system matching.

    Jamf reports serials in mixed case and Helix's asset intake accepts
    whatever the operator typed, so every serial comparison in this
    codebase lowercases + strips. Centralised here so the sync writer and
    every reader derive the key the same way — a mismatch between the two
    silently unlinks a device from its asset.
    """
    cleaned = (serial or "").strip().lower()
    return cleaned or None


class MdmManagedDevice(Base):
    """One device enrolled in an MDM tenant (currently Jamf Pro).

    Refreshed wholesale by the ``jamf.device_sync`` scheduled job: every
    run upserts the devices the tenant reports and marks the rows it no
    longer sees as ``gone``. Treat every resolution column below as a
    projection that the next sync may recompute — the authoritative
    facts are ``(integration_id, device_type, external_id)`` and the
    serial.

    **Why there are no foreign keys on the resolution columns.**
    ``asset_id`` / ``assigned_person_id`` / ``cleanup_task_id`` /
    ``integration_id`` are plain UUIDs on purpose. TiDB takes a lock on
    the PARENT row for every child INSERT that carries an FK, and this
    table is rewritten in a single nightly batch that would otherwise
    serialise ~50 inserts behind locks on one integration row and on
    whatever ``identity_persons`` rows the fleet resolves to — the exact
    03:00 cron-collision shape that has bitten this platform before.
    Readers LEFT JOIN and render a dangling id as "—"; the sync repairs
    stale ids on its next pass.
    """

    __tablename__ = "asset_mdm_devices"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        # The MDM's own primary key, scoped to the tenant. Device type is
        # part of the key because Jamf numbers computers and mobile
        # devices in separate sequences — id 42 is two different devices.
        UniqueConstraint(
            "integration_id", "device_type", "external_id",
            name="uq_asset_mdm_devices_external",
        ),
        # The asset-matching path: dispose an asset, look the device up
        # by serial. Not unique — a tenant can hold two records for one
        # serial after a re-enrolment, and Helix must show both rather
        # than pick one.
        Index("ix_asset_mdm_devices_serial", "serial_key"),
        Index("ix_asset_mdm_devices_asset", "asset_id"),
        Index("ix_asset_mdm_devices_person", "assigned_person_id"),
        # The two list views: "what needs cleaning up" and "what is
        # enrolled to a leaver".
        Index(
            "ix_asset_mdm_devices_state", "enrollment_state", "match_category",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )

    # ── identity in the MDM tenant ────────────────────────────────
    integration_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    # 'jamf' today. Carried explicitly so a second MDM (Intune) can share
    # the ledger without the readers having to resolve the integration.
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    # 'computer' | 'mobile' — the two Jamf inventories.
    device_type: Mapped[str] = mapped_column(String(16), nullable=False)
    # Jamf inventory id. String, not int: the API returns it as a string
    # and a future MDM may key on a GUID.
    external_id: Mapped[str] = mapped_column(String(64), nullable=False)
    # Jamf ``managementId`` — MDM commands key off this, not the
    # inventory id. Nullable: a record can exist before it checks in.
    management_id: Mapped[str | None] = mapped_column(String(64))

    # ── hardware facts ────────────────────────────────────────────
    serial_number: Mapped[str | None] = mapped_column(String(128))
    # Lowercased serial. Stored rather than computed so the serial lookup
    # is a plain indexed equality — TiDB's expression indexes are not a
    # dependency worth taking for a column we control on write.
    serial_key: Mapped[str | None] = mapped_column(String(128))
    device_name: Mapped[str | None] = mapped_column(String(255))
    model: Mapped[str | None] = mapped_column(String(255))
    os_version: Mapped[str | None] = mapped_column(String(64))
    # Jamf's own "is this device under management" flag. An unmanaged
    # record still occupies inventory and still needs removing on
    # disposal, so it is a display attribute, not a filter.
    managed: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default=text("1"),
    )

    # ── what the MDM says about ownership ─────────────────────────
    mdm_assigned_email: Mapped[str | None] = mapped_column(String(255))
    mdm_assigned_name: Mapped[str | None] = mapped_column(String(255))

    # ── what Helix resolved (recomputed every sync) ───────────────
    asset_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    assigned_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    # 'mdm' when the person came from the MDM's own assigned-user field,
    # 'asset' when Helix's asset ledger supplied it (Jamf's field was
    # blank). Keeps "Jamf knows who holds this" distinct from "we worked
    # it out" — the first is a fact about Jamf's data quality, and
    # collapsing them hides how much of the fleet Jamf cannot attribute.
    owner_source: Mapped[str | None] = mapped_column(String(16))
    match_category: Mapped[str] = mapped_column(
        String(24), nullable=False, default=MDM_MATCH_UNOWNED,
        server_default=text("'unowned'"),
    )

    # ── enrolment lifecycle ───────────────────────────────────────
    enrollment_state: Mapped[str] = mapped_column(
        String(24), nullable=False, default=MDM_STATE_ENROLLED,
        server_default=text("'enrolled'"),
    )
    # The provisioning task carrying the "remove this from Jamf" todo.
    # No FK — see the class docstring.
    cleanup_task_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    cleanup_requested_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # Why cleanup was asked for: the asset's ``disposal_type`` (sold /
    # scrapped / donated / lost …), prefixed ``buyback`` when the
    # disposal came from an approved offboarding buyback. Recorded so the
    # card can say what happened to the machine instead of "it is gone".
    cleanup_reason: Mapped[str | None] = mapped_column(String(64))
    cleanup_confirmed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # 'sync' when the device stopped appearing in the tenant (the vendor
    # confirmed it), 'admin' when a human ticked the card. Only the first
    # is proof; the second is a claim, and the ledger keeps them apart.
    cleanup_confirmed_via: Mapped[str | None] = mapped_column(String(16))
    # When the vendor's roster last overturned a human's "I removed it".
    # A durable fact, because the alternative was inferring it from
    # ``cleanup_confirmed_at`` — which the contradiction itself clears,
    # so every predicate built on that field fired during the grace
    # window (the one case that is NOT a contradiction) and never after.
    # Kept set until the cleanup actually resolves, so an operator can
    # see that this device has been claimed-and-refuted before.
    cleanup_contradicted_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # Last time the open cleanup was nudged. On the device rather than on
    # the task because it is the DEVICE that is still enrolled — the
    # reminder exists while the tenant keeps reporting it, and survives a
    # task being cancelled and re-raised.
    cleanup_last_reminded_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )

    # ── observation window ────────────────────────────────────────
    first_seen_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # Last sync run that saw this device in the tenant.
    last_seen_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # The device's own last check-in with the MDM (Jamf lastContactTime).
    last_contact_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    # First sync run that did NOT see it. Cleared if it comes back.
    gone_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(),
    )

    def __repr__(self) -> str:
        return (
            f"<MdmManagedDevice {self.provider}:{self.device_type}"
            f"/{self.external_id} {self.serial_number} "
            f"[{self.enrollment_state}]>"
        )

# ── 财务固定资产台账对账 (migration 605, 2026-09-05) ──────────────────

RECONCILE_RUN_PREVIEWING = "previewing"
RECONCILE_RUN_APPLIED = "applied"
RECONCILE_RUN_DISCARDED = "discarded"

# The register line matched exactly one Helix asset which has NO amount
# yet, or whose amount is an estimate the register supersedes.
RECONCILE_MATCH_MATCHED = "matched"
# Matched one asset, but Helix already RECORDS a different amount. Not an
# error and not something to overwrite silently — a real discrepancy
# between two books that a human should look at.
RECONCILE_MATCH_CONFLICT = "conflict"
# The register line matches no Helix asset. Finance is depreciating a
# device the IT ledger does not know about — the inverse of the
# completeness gap, and invisible from every other surface.
RECONCILE_MATCH_UNMATCHED = "unmatched"
# The serial matches MORE THAN ONE Helix asset. Never guessed: picking
# one would write a money value onto an arbitrary row.
RECONCILE_MATCH_AMBIGUOUS = "ambiguous"
# The register line itself is unusable (no serial, unparseable amount).
RECONCILE_MATCH_INVALID = "invalid"

RECONCILE_DECISION_PENDING = "pending"
RECONCILE_DECISION_APPLIED = "applied"
RECONCILE_DECISION_SKIPPED = "skipped"


class AssetCostReconcileRun(Base):
    """One upload of the finance fixed-asset register.

    Preview-then-apply: a run stays ``previewing`` until an operator
    accepts specific rows, so an abandoned upload looks abandoned instead
    of looking like it landed. Counts are denormalised for the list view
    (a run can carry thousands of items and the index page must not sum
    them per row).
    """

    __tablename__ = "asset_cost_reconcile_runs"
    # ``uploaded_at`` / ``updated_at`` are server defaults and the
    # serializer reads them immediately after flush. Without
    # eager_defaults SQLAlchemy lazy-loads them on first access, which
    # outside the greenlet is a MissingGreenlet — same reason
    # ``AssetLocation`` carries this.
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    source_label: Mapped[str] = mapped_column(String(255), nullable=False)
    file_name: Mapped[str | None] = mapped_column(String(255))
    uploaded_by: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    uploaded_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    status: Mapped[str] = mapped_column(
        String(16),
        nullable=False,
        default=RECONCILE_RUN_PREVIEWING,
        server_default=RECONCILE_RUN_PREVIEWING,
    )
    row_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    matched_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    conflict_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    unmatched_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    applied_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0"),
    )
    applied_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    applied_by: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    notes: Mapped[str | None] = mapped_column(Text)
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=func.now(),
        onupdate=func.now(),
    )


class AssetCostReconcileItem(Base):
    """One register line with its match verdict and Helix-side snapshot.

    ``current_*`` is snapshotted AT PREVIEW TIME on purpose: the operator
    made the apply decision against those values, so re-deriving them at
    apply time would hide a concurrent edit rather than surface it. The
    apply path compares snapshot against live and refuses the row if they
    have diverged.

    No FK to ``asset_items``: an unmatched line has no asset, and a
    matched asset can later be hard-deleted without invalidating the
    historical record of what the register said.
    """

    __tablename__ = "asset_cost_reconcile_items"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    run_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    serial_number: Mapped[str | None] = mapped_column(String(128))
    serial_key: Mapped[str | None] = mapped_column(String(128))
    asset_tag_hint: Mapped[str | None] = mapped_column(String(64))
    register_amount: Mapped[Decimal | None] = mapped_column(Numeric(14, 2))
    register_currency: Mapped[str | None] = mapped_column(String(8))
    register_purchase_date: Mapped[datetime | None] = mapped_column(Date)
    register_entity: Mapped[str | None] = mapped_column(String(128))
    register_row_number: Mapped[int | None] = mapped_column(Integer)
    asset_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    current_amount: Mapped[Decimal | None] = mapped_column(Numeric(14, 2))
    current_currency: Mapped[str | None] = mapped_column(String(8))
    current_cost_source: Mapped[str | None] = mapped_column(String(24))
    match_status: Mapped[str] = mapped_column(String(16), nullable=False)
    match_note: Mapped[str | None] = mapped_column(String(255))
    decision: Mapped[str] = mapped_column(
        String(16),
        nullable=False,
        default=RECONCILE_DECISION_PENDING,
        server_default=RECONCILE_DECISION_PENDING,
    )
    decided_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 SeatUsageSignal(Base):
    """Persisted "when did this person last actually use it?" per vendor.

    Written by ``software_seat.usage_signal_refresh`` from the per-vendor
    collectors in :mod:`app.integrations.usage_collectors`, read by
    ``software_seat.idle_reclaim``. One row per
    ``(person, provider, product_key)``.

    Why a table rather than the live fetch idle-reclaim used to do:

    * **Every vendor, one shape.** Atlassian was reachable only through
      its own bespoke directory; Google and Zoom had no reader at all, so
      their paid seats were reclaimable only by a human noticing.
    * **The evidence outlives the run.** ``last_active_at`` alone answers
      "is this idle"; ``state`` + ``evidence`` + ``collected_at`` answer
      "and how do you know", which is the half an operator needs before
      revoking somebody's access.

    ``state`` is the load-bearing column, not ``last_active_at``. See
    :mod:`app.integrations.usage_signals` — ``unknown`` (the vendor
    declined to answer) and ``unsupported`` (the vendor cannot answer)
    must never be read as "idle", and a row older than the freshness
    window resolves back to ``unknown`` no matter what it says.
    """

    __tablename__ = "asset_seat_usage_signals"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        UniqueConstraint(
            "person_id", "provider", "product_key",
            name="uq_seat_usage_signal_person_provider_product",
        ),
        Index("ix_seat_usage_signal_provider_product", "provider", "product_key"),
        Index("ix_seat_usage_signal_collected_at", "collected_at"),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False, index=True,
    )
    #: ``integrations.provider`` slug — ``jira`` / ``google_workspace`` /
    #: ``zoom``. Deliberately the integration's own vocabulary so the
    #: collector registry keys off exactly what the connector says.
    provider: Mapped[str] = mapped_column(String(64), nullable=False)
    #: Product family inside the vendor (``jira-software`` /
    #: ``confluence`` / ``workspace`` / ``zoom``). A person can be a daily
    #: Confluence user and never open Jira, so the signal cannot be
    #: per-vendor only.
    product_key: Mapped[str] = mapped_column(String(128), nullable=False)
    #: ``active`` / ``never`` / ``unknown`` / ``unsupported``.
    state: Mapped[str] = mapped_column(String(16), nullable=False)
    #: NULL for every state except ``active`` — including ``never``,
    #: whose idle clock runs from the seat grant instead.
    last_active_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    #: What the vendor actually said, kept so a disputed reclaim can be
    #: answered without re-querying: raw field value, account status,
    #: sites, org unit.
    evidence: Mapped[dict[str, Any] | None] = mapped_column(JSON)
    #: When the roster this row came from was read. The freshness gate
    #: keys off this; a row nobody refreshed cannot testify about today.
    collected_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False,
    )
    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(),
    )