"""Procurement domain service — CRUD + derived-field resolution + history.

Boundaries:

* This service owns reads/writes against the ``procurement_*`` tables.
* It JOINs to ``identity_persons``, ``asset_software_catalog``,
  ``asset_legal_entities`` for display fields only — no cross-domain
  mutation.
* Contract edits write to ``procurement_contract_versions`` BEFORE the
  main row is updated so an interrupted edit doesn't lose history.
* Auto-numbering (supplier_code / contract_number / order_number /
  payment_number) uses a transactional `(prefix, calendar year)` counter.
  The first allocation bootstraps from legacy rows; resource + counter commit
  or roll back together. DB UNIQUE remains the final duplicate arbiter.
"""
from __future__ import annotations

import hashlib
import json
import logging
import unicodedata
import uuid
from collections.abc import Awaitable, Callable
from datetime import UTC, date, datetime, timedelta
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from functools import wraps
from typing import Any, Never

from fastapi.encoders import jsonable_encoder
from sqlalchemy import and_, case, func, literal, or_, select, union_all
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUser
from app.core.exceptions import (
    AppException,
    IdempotencyKeyInvalidError,
    IdempotencyPayloadMismatchError,
    NotFoundError,
    ValidationError,
)
from app.core.fx_rate import resolve_cny_per_usd
from app.core.idempotency import IdempotencyService
from app.domains.asset.models import (
    AssetSoftwareCatalog,
)
from app.domains.procurement.models import (
    CONTRACT_STATUSES,
    CONTRACT_TYPES,
    PAYMENT_STATUSES,
    PO_ORDER_TYPES,
    PO_STATUSES,
    RECEIPT_SEMANTICS,
    RECEIPT_STATUSES,
    SUPPLIER_STATUSES,
    ProcurementCategory,
    ProcurementContract,
    ProcurementContractAttachment,
    ProcurementContractVersion,
    ProcurementOrder,
    ProcurementOrderItem,
    ProcurementPayment,
    ProcurementSerialCounter,
    ProcurementSupplier,
    SoftwareProcurementMeta,
)
from app.domains.procurement.schemas import (
    CategoryCreate,
    CategoryUpdate,
    ContractCreate,
    ContractUpdate,
    OrderCreate,
    OrderItemCreate,
    OrderItemUpdate,
    OrderUpdate,
    PaymentCreate,
    PaymentUpdate,
    SoftwareProcurementMetaUpdate,
    SupplierCreate,
    SupplierUpdate,
)
from app.identity.person_models import IdentityPerson as Person

logger = logging.getLogger(__name__)


def _raise_leaver_runtime_locked() -> Never:
    raise AppException(
        code="LEAVER_RUNTIME_LOCKED",
        message=(
            "Your account is in the offboarding process. "
            "You can view data but cannot perform this action."
        ),
        status_code=403,
    )


async def lock_active_actor_for_mutation(
    db: AsyncSession,
    *,
    actor: AuthUser | None,
) -> None:
    """Lock and validate the authoritative actor row for a write transaction.

    The actor object was loaded during authentication and may be stale by the
    time a procurement mutation begins.  A locking current read makes the
    procurement commit and an offboarding status update serialize on the same
    ``identity_persons`` row: whichever transaction obtains the lock first is
    the operation that happened first.  Selecting scalar columns avoids the
    ORM identity map returning the already-loaded, stale actor instance.
    """

    actor_id = getattr(actor, "id", None)
    if actor_id is None:
        _raise_leaver_runtime_locked()

    row = (
        await db.execute(
            select(Person.employment_status, Person.deleted_at)
            .where(Person.id == actor_id)
            # A shared locking current read is sufficient: concurrent writes
            # by the same active operator may proceed, while an offboarding
            # UPDATE needs the exclusive row lock and therefore waits for all
            # in-flight procurement transactions to finish.
            .with_for_update(read=True)
        )
    ).one_or_none()
    if row is None or row.employment_status != "active" or row.deleted_at is not None:
        _raise_leaver_runtime_locked()


def _active_actor_required[**P, R](
    fn: Callable[P, Awaitable[R]],
) -> Callable[..., Awaitable[R]]:
    """Fail closed unless a procurement write names an active human actor."""

    @wraps(fn)
    async def guarded(*args: Any, actor: AuthUser | None = None, **kwargs: Any) -> R:
        db = args[0] if args else kwargs.get("db")
        if not isinstance(db, AsyncSession):
            _raise_leaver_runtime_locked()
        await lock_active_actor_for_mutation(db, actor=actor)
        return await fn(*args, **kwargs)

    return guarded


# Cap "supports_legal_entities" updates so a runaway JSON payload can't
# blow up a supplier row. 50 distinct legal entities per supplier is
# already absurd for a 500-person company.
_MAX_SUPPORTS_LEGAL_ENTITIES = 50

# Anti-double-submit window for create_payment: a second create with an
# identical (order/contract, amount, currency, status) inside this many
# seconds is treated as a retry of the first and returns the existing row
# instead of booking a duplicate financial payment.
_PAYMENT_DEDUP_WINDOW_SECONDS = 60
_SERIAL_RETRY_ATTEMPTS = 64
_PAYMENT_MAX_AMOUNT = Decimal("999999999999.99")


def _canonical_payment_amount(value: Decimal | None) -> Decimal | None:
    if value is None:
        return None
    if not value.is_finite() or value < 0:
        raise ValidationError("amount must be a finite non-negative number")
    try:
        value = value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    except InvalidOperation as exc:
        raise ValidationError("amount is outside the supported range") from exc
    if value > _PAYMENT_MAX_AMOUNT:
        raise ValidationError(
            f"amount must not exceed {_PAYMENT_MAX_AMOUNT}"
        )
    return value


def _canonical_payment_text(
    value: str | None,
    *,
    field: str,
    max_length: int,
    uppercase: bool = False,
    required: bool = False,
) -> str | None:
    if value is None:
        if required:
            raise ValidationError(f"{field} must not be empty")
        return None
    canonical = unicodedata.normalize("NFKC", value).strip()
    if uppercase:
        canonical = canonical.upper()
    if not canonical:
        if required:
            raise ValidationError(f"{field} must not be empty")
        return None
    if len(canonical) > max_length:
        raise ValidationError(
            f"{field} must be at most {max_length} characters after normalization"
        )
    return canonical


def _payment_create_lock_key(
    payload: PaymentCreate,
    *,
    amount: Decimal | None,
    currency: str,
    invoice_number: str | None,
    payment_number: str | None,
) -> str:
    """Return a private, deterministic key for the complete payment intent."""
    lock_identity = json.dumps(
        {
            "amount": str(amount) if amount is not None else None,
            "currency": currency,
            "invoice_number": invoice_number,
            "payment_number": payment_number,
            "status": payload.status,
            "references": {
                "contract_id": (
                    str(payload.contract_id) if payload.contract_id else None
                ),
                "order_id": str(payload.order_id) if payload.order_id else None,
                "software_catalog_id": (
                    str(payload.software_catalog_id)
                    if payload.software_catalog_id else None
                ),
            },
        },
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    digest = hashlib.sha256(lock_identity.encode("utf-8")).hexdigest()
    return f"payment_create:{digest}"


# ──────────────────────────────────────────────────────────────────────
# Helpers
# ──────────────────────────────────────────────────────────────────────


def _procurement_create_fingerprint(payload: Any) -> str:
    """Canonical payload fingerprint shared by HTTP and MCP create paths."""
    encoded = json.dumps(
        jsonable_encoder(payload),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


async def _idempotent_create[CreateModel](
    db: AsyncSession,
    *,
    operation: str,
    actor_id: uuid.UUID,
    actor: AuthUser | None,
    idempotency_key: str | None,
    payload: Any,
    model: type[CreateModel],
    creator: Callable[[], Awaitable[CreateModel]],
) -> tuple[CreateModel, bool]:
    """Create one procurement resource under the platform intent record.

    The caller owns the commit.  Consequently the resource row, PO line
    items, budget lock, audit row, and completed idempotency record all land
    atomically.  ``bool`` is True for a replay so callers do not emit a
    second audit event.
    """
    await lock_active_actor_for_mutation(db, actor=actor)
    if actor is None or actor.id is None or str(actor.id) != str(actor_id):
        _raise_leaver_runtime_locked()

    key = (idempotency_key or "").strip() or None
    if key is None:
        return await creator(), False
    if len(key) > 128 or not key.isprintable():
        raise IdempotencyKeyInvalidError(
            "Idempotency-Key must be 1-128 printable characters",
        )

    idem = IdempotencyService(
        db,
        scope=f"procurement.{operation}.create:{actor_id}",
        key=key,
        request_fingerprint=_procurement_create_fingerprint(payload),
    )
    try:
        cached = await idem.lookup()
    except IdempotencyPayloadMismatchError as exc:
        # #2284 acceptance contract is 409 for one key identifying a
        # different procurement intent (rather than the platform default
        # 422 used by generic state-transition endpoints).
        raise AppException(
            code="PROCUREMENT_IDEMPOTENCY_PAYLOAD_MISMATCH",
            message=exc.message,
            status_code=409,
        ) from exc

    if cached is not None:
        resource_id = cached.get("resource_id")
        if not resource_id:
            raise RuntimeError(
                f"completed procurement {operation} intent has no resource_id"
            )
        existing = await db.get(model, str(resource_id))
        if existing is None:
            raise RuntimeError(
                f"completed procurement {operation} intent points to a missing row"
            )
        return existing, True

    try:
        created = await creator()
        resource_id = getattr(created, "id")
        await idem.record(
            {"resource_id": str(resource_id)},
            resource_id=resource_id,
            status_code=200,
        )
        return created, False
    except Exception:
        await idem.mark_failed()
        raise


def _validate_in(value: str | None, valid_set: set[str] | frozenset[str], field: str) -> None:
    if value is None:
        return
    if value not in valid_set:
        raise ValidationError(
            f"{field} must be one of {sorted(valid_set)}; got {value!r}"
        )


async def _person_name_map(
    db: AsyncSession, person_ids: list[uuid.UUID | None]
) -> dict[uuid.UUID, str]:
    clean = [pid for pid in person_ids if pid is not None]
    if not clean:
        return {}
    # asyncmy UUID → BINARY bind quirk: cast to str at the boundary
    # (memory: asyncmy_uuid_bind). person.id is VARCHAR(36) on
    # production schemas.
    str_ids = [str(p) for p in clean]
    rows = await db.execute(
        select(Person.id, Person.display_name).where(Person.id.in_(str_ids))
    )
    return {uuid.UUID(str(row[0])): (row[1] or "") for row in rows}


async def _existing_max_serial(
    db: AsyncSession,
    prefix: str,
    table_col_pairs: list[tuple[Any, Any]],
) -> int:
    """Bootstrap a new counter from already-issued legacy numbers."""
    year = datetime.now(UTC).year
    pattern = f"{prefix}-{year}-%"
    max_seq = 0
    for _model, col in table_col_pairs:
        result = await db.execute(select(col).where(col.like(pattern)))
        for (val,) in result:
            try:
                suffix = int(val.rsplit("-", 1)[-1])
                max_seq = max(max_seq, suffix)
            except (ValueError, AttributeError):
                continue
    return max_seq


async def _allocate_serial(
    db: AsyncSession,
    prefix: str,
    model: Any,
    column: Any,
) -> str:
    """Atomically allocate a human number from the transactional counter.

    The initial row self-bootstraps from existing resource numbers. Concurrent
    first callers may compute the same bootstrap value, but the composite PK
    turns the second INSERT into ``last_value + 1``. The UPSERT row lock stays
    held until the resource transaction commits or rolls back.
    """
    year = datetime.now(UTC).year
    existing = await db.scalar(
        select(ProcurementSerialCounter.last_value).where(
            ProcurementSerialCounter.prefix == prefix,
            ProcurementSerialCounter.serial_year == year,
        )
    )
    bootstrap = existing
    if bootstrap is None:
        bootstrap = await _existing_max_serial(db, prefix, [(model, column)])

    stmt = mysql_insert(ProcurementSerialCounter).values(
        prefix=prefix,
        serial_year=year,
        last_value=int(bootstrap) + 1,
    )
    stmt = stmt.on_duplicate_key_update(
        last_value=ProcurementSerialCounter.last_value + 1,
        updated_at=func.now(),
    )
    await db.execute(stmt)
    allocated = await db.scalar(
        select(ProcurementSerialCounter.last_value)
        .where(
            ProcurementSerialCounter.prefix == prefix,
            ProcurementSerialCounter.serial_year == year,
        )
        .with_for_update()
    )
    if allocated is None:
        raise RuntimeError(f"procurement counter {prefix}/{year} disappeared")
    return f"{prefix}-{year}-{int(allocated):05d}"


async def _advance_serial_floor(
    db: AsyncSession,
    prefix: str,
    explicit_number: str,
) -> None:
    """Advance the counter when a caller supplies a higher canonical number."""
    parts = explicit_number.split("-")
    if len(parts) != 3 or parts[0] != prefix:
        return
    try:
        year = int(parts[1])
        suffix = int(parts[2])
    except ValueError:
        return
    if year < 1 or suffix < 1:
        return
    stmt = mysql_insert(ProcurementSerialCounter).values(
        prefix=prefix,
        serial_year=year,
        last_value=suffix,
    )
    await db.execute(stmt.on_duplicate_key_update(
        last_value=func.greatest(ProcurementSerialCounter.last_value, suffix),
        updated_at=func.now(),
    ))


def _is_serial_collision(exc: IntegrityError, serial_column: str) -> bool:
    """Match only MySQL/TiDB 1062 for the intended serial unique key."""
    orig = getattr(exc, "orig", None)
    args = getattr(orig, "args", None)
    if not args or args[0] != 1062:
        return False
    return serial_column.lower() in str(orig).lower()


async def _insert_numbered[SerialModel](
    db: AsyncSession,
    *,
    prefix: str,
    model: Any,
    column: Any,
    serial_column: str,
    explicit_number: str | None,
    build: Callable[[str], SerialModel],
) -> SerialModel:
    """Insert one numbered row without poisoning the caller transaction.

    UNIQUE is the cross-node arbiter. Each candidate INSERT lives in a
    SAVEPOINT; an auto-number collision is regenerated from a locking current
    read, while an explicit duplicate becomes a typed 409. Any unrelated FK
    or unique error still surfaces unchanged.
    """
    for _attempt in range(_SERIAL_RETRY_ATTEMPTS):
        number = explicit_number or await _allocate_serial(db, prefix, model, column)
        candidate = build(number)
        try:
            # Add inside the SAVEPOINT so rollback expunges a failed candidate
            # instead of leaving it in session.new for the next autoflush.
            async with db.begin_nested():
                # Keep one lock order for canonical explicit and automatic
                # numbers: counter row first, target UNIQUE index second. If
                # an explicit ``PREFIX-YEAR-N`` row were inserted first and
                # only then advanced the counter, it could deadlock with an
                # automatic allocator that already held the counter and was
                # waiting to insert the same N. The SAVEPOINT also ensures a
                # rejected explicit duplicate cannot advance the durable
                # floor or create an attacker-controlled numbering hole.
                if explicit_number is not None:
                    await _advance_serial_floor(db, prefix, explicit_number)
                db.add(candidate)
                await db.flush()
        except IntegrityError as exc:
            if not _is_serial_collision(exc, serial_column):
                raise
            if explicit_number is not None:
                raise AppException(
                    code="PROCUREMENT_SERIAL_CONFLICT",
                    message=f"{serial_column} {explicit_number!r} already exists",
                    status_code=409,
                ) from exc
            # A stale/missing counter met a legacy/direct-SQL number. Advance
            # to the actual table maximum and retry without poisoning the txn.
            floor = await _existing_max_serial(db, prefix, [(model, column)])
            await _advance_serial_floor(
                db, prefix, f"{prefix}-{datetime.now(UTC).year}-{floor}",
            )
            continue
        return candidate
    raise AppException(
        code="PROCUREMENT_SERIAL_CONTENTION",
        message=f"Could not allocate {prefix} number after concurrent retries",
        status_code=503,
    )


async def _resolve_fx(
    db: AsyncSession, amount: Decimal | None, currency: str | None
) -> tuple[Decimal | None, Decimal | None]:
    """Return ``(fx_rate_to_cny, amount_cny)`` for the given (amount, currency).

    CNY → 1.0 / amount unchanged. USD → ``resolve_cny_per_usd``. Other
    currencies (SGD / JPY / HKD / EUR…) are passed through with NULL
    cny — Phase 1 does not pull rates for them; PR 3 reminders that
    care about CNY rollups will skip these rows.
    """
    if amount is None or currency is None:
        return None, None
    cur = currency.upper()
    if cur == "CNY":
        return Decimal("1.000000"), amount
    if cur == "USD":
        try:
            rate = await resolve_cny_per_usd(db)
        except RuntimeError:
            # Rate unavailable; leave NULL — PR 3 reminders that care
            # about CNY rollups raise their own alert.
            return None, None
        d_rate = Decimal(str(rate)).quantize(Decimal("0.000001"))
        return d_rate, (amount * d_rate).quantize(Decimal("0.01"))
    return None, None


# ──────────────────────────────────────────────────────────────────────
# Categories
# ──────────────────────────────────────────────────────────────────────


async def list_categories(
    db: AsyncSession, *, include_inactive: bool = False
) -> list[ProcurementCategory]:
    q = select(ProcurementCategory).order_by(
        ProcurementCategory.sort_order.asc(), ProcurementCategory.name_zh.asc()
    )
    if not include_inactive:
        q = q.where(ProcurementCategory.is_active.is_(True))
    return list((await db.execute(q)).scalars())


@_active_actor_required
async def create_category(
    db: AsyncSession, payload: CategoryCreate
) -> ProcurementCategory:
    _validate_in(payload.receipt_semantics, RECEIPT_SEMANTICS, "receipt_semantics")
    existing = await db.execute(
        select(ProcurementCategory).where(ProcurementCategory.code == payload.code)
    )
    if existing.scalar_one_or_none() is not None:
        raise ValidationError(f"Category code {payload.code!r} already exists")
    obj = ProcurementCategory(**payload.model_dump())
    db.add(obj)
    await db.flush()
    return obj


@_active_actor_required
async def update_category(
    db: AsyncSession, category_id: uuid.UUID, payload: CategoryUpdate
) -> ProcurementCategory:
    obj = await db.get(ProcurementCategory, str(category_id))
    if obj is None:
        raise NotFoundError(f"Category {category_id} not found")
    data = payload.model_dump(exclude_unset=True)
    _validate_in(data.get("receipt_semantics"), RECEIPT_SEMANTICS, "receipt_semantics")
    for k, v in data.items():
        setattr(obj, k, v)
    await db.flush()
    return obj


@_active_actor_required
async def delete_category(db: AsyncSession, category_id: uuid.UUID) -> None:
    obj = await db.get(ProcurementCategory, str(category_id))
    if obj is None:
        raise NotFoundError(f"Category {category_id} not found")
    # Refuse delete if referenced — soft-deactivate via is_active=False
    # instead, otherwise downstream display columns go blank. category_id
    # is an FK from orders, contracts AND budgets; all three must be
    # checked or the DELETE either raises an opaque FK IntegrityError
    # (RESTRICT, no ondelete) or orphans the referrers.
    from app.domains.procurement.models import ProcurementBudget
    order_refs = await db.execute(
        select(func.count(ProcurementOrder.id)).where(
            ProcurementOrder.category_id == obj.id
        )
    )
    contract_refs = await db.execute(
        select(func.count(ProcurementContract.id)).where(
            ProcurementContract.category_id == obj.id
        )
    )
    budget_refs = await db.execute(
        select(func.count(ProcurementBudget.id)).where(
            ProcurementBudget.category_id == obj.id
        )
    )
    if (
        (order_refs.scalar() or 0)
        or (contract_refs.scalar() or 0)
        or (budget_refs.scalar() or 0)
    ):
        raise ValidationError(
            "Category is referenced by procurement orders, contracts or "
            "budgets; deactivate instead of deleting."
        )
    await db.delete(obj)


# ──────────────────────────────────────────────────────────────────────
# Suppliers
# ──────────────────────────────────────────────────────────────────────


async def list_suppliers(
    db: AsyncSession,
    *,
    keyword: str | None = None,
    status: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[ProcurementSupplier], int]:
    base = select(ProcurementSupplier)
    if keyword:
        like = f"%{keyword}%"
        base = base.where(
            or_(
                ProcurementSupplier.name.like(like),
                ProcurementSupplier.legal_name.like(like),
                ProcurementSupplier.supplier_code.like(like),
            )
        )
    if status:
        _validate_in(status, SUPPLIER_STATUSES, "status")
        base = base.where(ProcurementSupplier.status == status)
    total = await db.scalar(select(func.count()).select_from(base.subquery()))
    rows = await db.execute(
        base.order_by(ProcurementSupplier.name.asc()).limit(limit).offset(offset)
    )
    return list(rows.scalars()), int(total or 0)


async def get_supplier(
    db: AsyncSession, supplier_id: uuid.UUID
) -> ProcurementSupplier:
    obj = await db.get(ProcurementSupplier, str(supplier_id))
    if obj is None:
        raise NotFoundError(f"Supplier {supplier_id} not found")
    return obj


async def _supplier_derived_counts(
    db: AsyncSession, supplier_id: uuid.UUID
) -> tuple[int, int, int]:
    """Return (contract_count, active_contract_count, po_count)."""
    return (await supplier_derived_counts(db, [supplier_id]))[supplier_id]


async def supplier_derived_counts(
    db: AsyncSession,
    supplier_ids: list[uuid.UUID],
) -> dict[uuid.UUID, tuple[int, int, int]]:
    """Return supplier card counts for a whole page in one query.

    Contracts and orders are aggregated independently before the results are
    unioned.  This avoids both the HTTP N+1 and the row multiplication that a
    three-way ``supplier -> contracts -> orders`` outer join would create.
    Missing parents are filled with zero without issuing per-row queries.
    """
    ids = list(dict.fromkeys(supplier_ids))
    if not ids:
        return {}
    str_ids = [str(value) for value in ids]
    contract_counts = (
        select(
            ProcurementContract.supplier_id.label("parent_id"),
            func.count(ProcurementContract.id).label("contract_count"),
            func.sum(case(
                (ProcurementContract.status == "active", 1), else_=0,
            )).label("active_contract_count"),
            literal(0).label("po_count"),
        )
        .where(ProcurementContract.supplier_id.in_(str_ids))
        .group_by(ProcurementContract.supplier_id)
    )
    order_counts = (
        select(
            ProcurementOrder.supplier_id.label("parent_id"),
            literal(0).label("contract_count"),
            literal(0).label("active_contract_count"),
            func.count(ProcurementOrder.id).label("po_count"),
        )
        .where(ProcurementOrder.supplier_id.in_(str_ids))
        .group_by(ProcurementOrder.supplier_id)
    )
    combined = union_all(contract_counts, order_counts).subquery()
    rows = await db.execute(
        select(
            combined.c.parent_id,
            func.sum(combined.c.contract_count),
            func.sum(combined.c.active_contract_count),
            func.sum(combined.c.po_count),
        ).group_by(combined.c.parent_id)
    )
    result = {value: (0, 0, 0) for value in ids}
    for parent_id, contracts, active, orders in rows:
        result[uuid.UUID(str(parent_id))] = (
            int(contracts or 0), int(active or 0), int(orders or 0),
        )
    return result


async def contract_derived_counts(
    db: AsyncSession,
    contract_ids: list[uuid.UUID],
) -> dict[uuid.UUID, tuple[int, int]]:
    """Return ``(attachment_count, po_count)`` for a page in one query."""
    ids = list(dict.fromkeys(contract_ids))
    if not ids:
        return {}
    str_ids = [str(value) for value in ids]
    attachment_counts = (
        select(
            ProcurementContractAttachment.contract_id.label("parent_id"),
            func.count(ProcurementContractAttachment.id).label("attachment_count"),
            literal(0).label("po_count"),
        )
        .where(ProcurementContractAttachment.contract_id.in_(str_ids))
        .group_by(ProcurementContractAttachment.contract_id)
    )
    order_counts = (
        select(
            ProcurementOrder.contract_id.label("parent_id"),
            literal(0).label("attachment_count"),
            func.count(ProcurementOrder.id).label("po_count"),
        )
        .where(ProcurementOrder.contract_id.in_(str_ids))
        .group_by(ProcurementOrder.contract_id)
    )
    combined = union_all(attachment_counts, order_counts).subquery()
    rows = await db.execute(
        select(
            combined.c.parent_id,
            func.sum(combined.c.attachment_count),
            func.sum(combined.c.po_count),
        ).group_by(combined.c.parent_id)
    )
    result = {value: (0, 0) for value in ids}
    for parent_id, attachments, orders in rows:
        result[uuid.UUID(str(parent_id))] = (
            int(attachments or 0), int(orders or 0),
        )
    return result


async def order_item_counts(
    db: AsyncSession,
    order_ids: list[uuid.UUID],
) -> dict[uuid.UUID, int]:
    """Return line-item counts for a page with one grouped query."""
    ids = list(dict.fromkeys(order_ids))
    if not ids:
        return {}
    rows = await db.execute(
        select(
            ProcurementOrderItem.order_id,
            func.count(ProcurementOrderItem.id),
        )
        .where(ProcurementOrderItem.order_id.in_([str(value) for value in ids]))
        .group_by(ProcurementOrderItem.order_id)
    )
    result = {value: 0 for value in ids}
    for order_id, count in rows:
        result[uuid.UUID(str(order_id))] = int(count or 0)
    return result


def _enforce_contacts_invariant(contacts: list[dict[str, Any]] | None) -> list[dict[str, Any]]:
    """Ensure at most one ``is_primary=True``; promote first row if none."""
    contacts = list(contacts or [])
    primaries = [c for c in contacts if c.get("is_primary")]
    if len(primaries) > 1:
        # Keep the first; demote the rest.
        first_seen = False
        for c in contacts:
            if c.get("is_primary"):
                if first_seen:
                    c["is_primary"] = False
                else:
                    first_seen = True
    elif not primaries and contacts:
        contacts[0]["is_primary"] = True
    return contacts


@_active_actor_required
async def create_supplier(
    db: AsyncSession, payload: SupplierCreate
) -> ProcurementSupplier:
    _validate_in(payload.status, SUPPLIER_STATUSES, "status")
    data = payload.model_dump()
    code = data.pop("supplier_code") or None

    contacts = _enforce_contacts_invariant(data.pop("contacts", None) or [])
    sle = list(data.pop("supports_legal_entities", None) or [])[
        : _MAX_SUPPORTS_LEGAL_ENTITIES
    ]
    return await _insert_numbered(
        db,
        prefix="SUP",
        model=ProcurementSupplier,
        column=ProcurementSupplier.supplier_code,
        serial_column="supplier_code",
        explicit_number=code,
        build=lambda number: ProcurementSupplier(
            supplier_code=number,
            contacts=contacts or None,
            supports_legal_entities=[str(x) for x in sle] or None,
            **data,
        ),
    )


@_active_actor_required
async def update_supplier(
    db: AsyncSession, supplier_id: uuid.UUID, payload: SupplierUpdate
) -> ProcurementSupplier:
    obj = await get_supplier(db, supplier_id)
    data = payload.model_dump(exclude_unset=True)
    _validate_in(data.get("status"), SUPPLIER_STATUSES, "status")
    if "contacts" in data and data["contacts"] is not None:
        data["contacts"] = _enforce_contacts_invariant(data["contacts"]) or None
    if "supports_legal_entities" in data and data["supports_legal_entities"] is not None:
        data["supports_legal_entities"] = [
            str(x) for x in data["supports_legal_entities"][:_MAX_SUPPORTS_LEGAL_ENTITIES]
        ] or None
    for k, v in data.items():
        setattr(obj, k, v)
    await db.flush()
    return obj


# ──────────────────────────────────────────────────────────────────────
# Contracts
# ──────────────────────────────────────────────────────────────────────


# Fields whose changes get written into procurement_contract_versions.
_CONTRACT_AUDITED_FIELDS = frozenset({
    "title", "supplier_id", "legal_entity_id", "category_id",
    "contract_type", "parent_contract_id",
    "total_amount", "currency",
    "signed_date", "effective_date", "expiry_date",
    "auto_renew", "renew_notice_days", "status",
    "business_owner_id", "it_contact_id", "finance_contact_id",
    "procurement_owner_id", "external_contract_no",
})


async def list_contracts(
    db: AsyncSession,
    *,
    keyword: str | None = None,
    supplier_id: uuid.UUID | None = None,
    status: str | None = None,
    category_id: uuid.UUID | None = None,
    expiring_within_days: int | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[ProcurementContract], int]:
    base = select(ProcurementContract)
    if keyword:
        like = f"%{keyword}%"
        base = base.where(
            or_(
                ProcurementContract.title.like(like),
                ProcurementContract.contract_number.like(like),
                ProcurementContract.external_contract_no.like(like),
            )
        )
    if supplier_id is not None:
        base = base.where(ProcurementContract.supplier_id == str(supplier_id))
    if status:
        _validate_in(status, CONTRACT_STATUSES, "status")
        base = base.where(ProcurementContract.status == status)
    if category_id is not None:
        base = base.where(ProcurementContract.category_id == str(category_id))
    if expiring_within_days is not None and expiring_within_days >= 0:
        today = date.today()
        from datetime import timedelta
        cutoff = today + timedelta(days=expiring_within_days)
        base = base.where(
            and_(
                ProcurementContract.expiry_date.is_not(None),
                ProcurementContract.expiry_date <= cutoff,
            )
        )
    total = await db.scalar(select(func.count()).select_from(base.subquery()))
    rows = await db.execute(
        base.order_by(ProcurementContract.created_at.desc())
        .limit(limit).offset(offset)
    )
    return list(rows.scalars()), int(total or 0)


async def get_contract(
    db: AsyncSession, contract_id: uuid.UUID, *, lock: bool = False,
) -> ProcurementContract:
    stmt = select(ProcurementContract).where(
        ProcurementContract.id == str(contract_id)
    )
    if lock:
        stmt = stmt.with_for_update()
    obj = await db.scalar(stmt)
    if obj is None:
        raise NotFoundError(f"Contract {contract_id} not found")
    return obj


async def list_contract_attachments(
    db: AsyncSession, contract_id: uuid.UUID,
) -> list[ProcurementContractAttachment]:
    rows = await db.execute(
        select(ProcurementContractAttachment)
        .where(ProcurementContractAttachment.contract_id == str(contract_id))
        .order_by(ProcurementContractAttachment.uploaded_at.desc())
    )
    return list(rows.scalars())


async def list_contract_versions(
    db: AsyncSession, contract_id: uuid.UUID,
) -> list[ProcurementContractVersion]:
    rows = await db.execute(
        select(ProcurementContractVersion)
        .where(ProcurementContractVersion.contract_id == str(contract_id))
        .order_by(ProcurementContractVersion.version_no.desc())
    )
    return list(rows.scalars())


@_active_actor_required
async def create_contract(
    db: AsyncSession, payload: ContractCreate
) -> ProcurementContract:
    _validate_in(payload.status, CONTRACT_STATUSES, "status")
    _validate_in(payload.contract_type, CONTRACT_TYPES, "contract_type")
    # SOW / amendment must have a parent.
    if payload.contract_type in ("sow", "amendment") and payload.parent_contract_id is None:
        raise ValidationError(
            f"contract_type={payload.contract_type!r} requires parent_contract_id"
        )
    data = payload.model_dump()
    number = data.pop("contract_number") or None
    return await _insert_numbered(
        db,
        prefix="CT",
        model=ProcurementContract,
        column=ProcurementContract.contract_number,
        serial_column="contract_number",
        explicit_number=number,
        build=lambda allocated: ProcurementContract(
            contract_number=allocated, **data,
        ),
    )


@_active_actor_required
async def update_contract(
    db: AsyncSession,
    contract_id: uuid.UUID,
    payload: ContractUpdate,
    *,
    actor_person_id: uuid.UUID | None = None,
) -> ProcurementContract:
    # Serialize every version allocator on the contract parent. The lock is
    # held by the caller transaction through its version INSERT + contract
    # update commit, so concurrent edits cannot observe the same last version.
    obj = await get_contract(db, contract_id, lock=True)
    data = payload.model_dump(exclude_unset=True)
    summary = data.pop("change_summary", None)
    _validate_in(data.get("status"), CONTRACT_STATUSES, "status")
    _validate_in(data.get("contract_type"), CONTRACT_TYPES, "contract_type")

    # Build the change manifest BEFORE updating.
    changed: dict[str, dict[str, Any]] = {}
    for field in _CONTRACT_AUDITED_FIELDS:
        if field not in data:
            continue
        new_val = data[field]
        old_val = getattr(obj, field, None)
        if new_val != old_val:
            changed[field] = {
                "from": _jsonable(old_val),
                "to": _jsonable(new_val),
            }

    if changed:
        # Find the next version number.
        next_v = await db.scalar(
            select(ProcurementContractVersion.version_no)
            .where(ProcurementContractVersion.contract_id == str(obj.id))
            .order_by(ProcurementContractVersion.version_no.desc())
            .limit(1)
            .with_for_update()
        ) or 0
        db.add(
            ProcurementContractVersion(
                contract_id=obj.id,
                version_no=int(next_v) + 1,
                changed_fields=changed,
                change_summary=summary,
                changed_by_id=actor_person_id,
            )
        )

    for k, v in data.items():
        setattr(obj, k, v)
    await db.flush()
    return obj


def _jsonable(value: Any) -> Any:
    """Recursively render a value into JSON-safe primitives."""
    if isinstance(value, uuid.UUID):
        return str(value)
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Decimal):
        return str(value)
    return value


# ──────────────────────────────────────────────────────────────────────
# Purchase orders
# ──────────────────────────────────────────────────────────────────────


async def list_orders(
    db: AsyncSession,
    *,
    keyword: str | None = None,
    supplier_id: uuid.UUID | None = None,
    contract_id: uuid.UUID | None = None,
    category_id: uuid.UUID | None = None,
    status: str | None = None,
    order_type: str | None = None,
    applicant_id: uuid.UUID | None = None,
    start_date: date | None = None,
    end_date: date | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[ProcurementOrder], int]:
    base = select(ProcurementOrder)
    if keyword:
        like = f"%{keyword}%"
        base = base.where(
            or_(
                ProcurementOrder.order_number.like(like),
                ProcurementOrder.external_po_number.like(like),
                ProcurementOrder.notes.like(like),
            )
        )
    if supplier_id is not None:
        base = base.where(ProcurementOrder.supplier_id == str(supplier_id))
    if contract_id is not None:
        base = base.where(ProcurementOrder.contract_id == str(contract_id))
    if category_id is not None:
        base = base.where(ProcurementOrder.category_id == str(category_id))
    if status:
        _validate_in(status, PO_STATUSES, "status")
        base = base.where(ProcurementOrder.status == status)
    if order_type:
        _validate_in(order_type, PO_ORDER_TYPES, "order_type")
        base = base.where(ProcurementOrder.order_type == order_type)
    if applicant_id is not None:
        base = base.where(ProcurementOrder.applicant_id == str(applicant_id))
    if start_date is not None:
        base = base.where(ProcurementOrder.created_at >= start_date)
    if end_date is not None:
        from datetime import timedelta
        base = base.where(
            ProcurementOrder.created_at < datetime.combine(
                end_date + timedelta(days=1), datetime.min.time()
            )
        )

    total = await db.scalar(select(func.count()).select_from(base.subquery()))
    rows = await db.execute(
        base.order_by(ProcurementOrder.created_at.desc())
        .limit(limit).offset(offset)
    )
    return list(rows.scalars()), int(total or 0)


async def get_order(
    db: AsyncSession, order_id: uuid.UUID, *, lock: bool = False,
) -> ProcurementOrder:
    stmt = select(ProcurementOrder).where(ProcurementOrder.id == str(order_id))
    if lock:
        stmt = stmt.with_for_update()
    obj = await db.scalar(stmt)
    if obj is None:
        raise NotFoundError(f"Order {order_id} not found")
    return obj


async def list_order_items(
    db: AsyncSession, order_id: uuid.UUID
) -> list[ProcurementOrderItem]:
    rows = await db.execute(
        select(ProcurementOrderItem)
        .where(ProcurementOrderItem.order_id == str(order_id))
        .order_by(ProcurementOrderItem.line_no.asc())
    )
    return list(rows.scalars())


@_active_actor_required
async def create_order(
    db: AsyncSession, payload: OrderCreate, *, applicant_id: uuid.UUID
) -> ProcurementOrder:
    _validate_in(payload.status, PO_STATUSES, "status")
    _validate_in(payload.order_type, PO_ORDER_TYPES, "order_type")

    data = payload.model_dump()
    items: list[dict[str, Any]] = data.pop("items", []) or []
    order_number = data.pop("order_number") or None

    # Resolve line numbers before allocating a PO number or writing anything.
    # ``enumerate`` alone is unsafe when explicit and implicit values mix
    # (for example explicit 2 followed by an omitted value at list position
    # 2), and duplicate explicit values otherwise surface as a raw 1062 that
    # poisons the session. Omitted lines take the smallest available positive
    # integer while explicit values remain stable.
    used_line_numbers: set[int] = set()
    for raw in items:
        explicit_line = raw.get("line_no")
        if explicit_line is not None:
            line_no = int(explicit_line)
            if line_no in used_line_numbers:
                raise ValidationError(
                    f"Duplicate line_no {line_no} in order items"
                )
            used_line_numbers.add(line_no)

    assigned_line_numbers: list[int] = []
    next_auto_line = 1
    for raw in items:
        explicit_line = raw.get("line_no")
        if explicit_line is None:
            while next_auto_line in used_line_numbers:
                next_auto_line += 1
            line_no = next_auto_line
            used_line_numbers.add(line_no)
            next_auto_line += 1
        else:
            line_no = int(explicit_line)
        assigned_line_numbers.append(line_no)

    # Applicant fallback: route param wins over payload.
    data.pop("applicant_id", None)

    # Roll up ``total_amount`` from line items when the caller omitted it,
    # so an items-only PO (the documented MCP create path) still has a
    # total — without one, ``_resolve_fx`` yields NULL ``total_cny`` and
    # ``lock_budget_for_order`` short-circuits, silently consuming zero
    # budget against the pool (PR 3's headline feature). Currency-safe:
    # we only sum when every item resolves to a concrete line amount AND
    # they all share the order's currency (an item ``currency`` of None
    # inherits the order currency). On any currency mix or missing amount
    # we leave ``total_amount`` NULL rather than add across currencies —
    # the budget pool then stays untouched, identical to prior behaviour.
    if data.get("total_amount") is None and items:
        order_ccy = data.get("currency") or "CNY"
        resolved: list[Decimal] = []
        consistent = True
        for raw in items:
            line_total = raw.get("line_total")
            if line_total is None:
                unit_price = raw.get("unit_price")
                quantity = raw.get("quantity")
                if unit_price is not None and quantity is not None:
                    line_total = Decimal(str(unit_price)) * Decimal(str(quantity))
            item_ccy = raw.get("currency") or order_ccy
            if line_total is None or item_ccy != order_ccy:
                consistent = False
                break
            resolved.append(Decimal(str(line_total)))
        if consistent and resolved:
            data["total_amount"] = sum(resolved, Decimal(0))

    fx_rate, total_cny = await _resolve_fx(
        db, data.get("total_amount"), data.get("currency")
    )

    obj = await _insert_numbered(
        db,
        prefix="PO",
        model=ProcurementOrder,
        column=ProcurementOrder.order_number,
        serial_column="order_number",
        explicit_number=order_number,
        build=lambda allocated: ProcurementOrder(
            order_number=allocated,
            applicant_id=applicant_id,
            fx_rate_to_cny=fx_rate,
            total_cny=total_cny,
            **data,
        ),
    )

    # Persist line items with auto line_no.
    for raw, line_no in zip(items, assigned_line_numbers, strict=True):
        # Preserve an explicit quantity — including a legitimate 0 — instead of
        # ``or Decimal(1)`` silently coercing 0 → 1 (which would inflate the
        # persisted line and desync it from the order total). Default to 1 only
        # when the caller omitted quantity entirely.
        raw_qty = raw.get("quantity")
        qty = Decimal(1) if raw_qty is None else raw_qty
        # Persist a derived line_total when the caller omitted it but the
        # line is computable (unit_price × quantity) — mirrors the total_amount
        # rollup above, so the line breakdown is complete instead of a NULL row
        # under a non-NULL order total.
        raw_line_total = raw.get("line_total")
        if raw_line_total is None and raw.get("unit_price") is not None:
            raw_line_total = Decimal(str(raw["unit_price"])) * Decimal(str(qty))
        line = ProcurementOrderItem(
            order_id=obj.id,
            line_no=line_no,
            description=raw["description"],
            quantity=qty,
            unit=raw.get("unit"),
            unit_price=raw.get("unit_price"),
            line_total=raw_line_total,
            currency=raw.get("currency") or obj.currency,
            target_asset_id=raw.get("target_asset_id"),
            target_software_id=raw.get("target_software_id"),
            target_license_plan_id=raw.get("target_license_plan_id"),
            target_descriptor=raw.get("target_descriptor"),
            notes=raw.get("notes"),
        )
        db.add(line)
    await db.flush()

    # PR 3: lock budget if a matching pool exists. Best-effort — no
    # match means no lock, no error. Per user direction 2026-05-30
    # ("Helix 创建 PO 就立刻预占预算"), this runs at creation regardless
    # of PO status (draft included) so the budget pool reflects in-
    # flight commitments early.
    # SAVEPOINT so a statement-level failure inside the budget lock
    # (IntegrityError / deadlock / FK) rolls back ONLY the nested scope and
    # leaves the outer session usable for the router's audit + commit. A
    # bare except with no rollback would poison the session and silently
    # drop the whole PO on the subsequent commit (PendingRollbackError).
    try:
        from app.domains.procurement.budget_service import lock_budget_for_order
        async with db.begin_nested():
            await lock_budget_for_order(db, obj, actor_person_id=applicant_id)
    except Exception:  # noqa: BLE001
        logger.exception("budget lock failed for PO %s; PO is unaffected", obj.order_number)

    return obj


@_active_actor_required
async def update_order(
    db: AsyncSession, order_id: uuid.UUID, payload: OrderUpdate
) -> ProcurementOrder:
    obj = await get_order(db, order_id)
    data = payload.model_dump(exclude_unset=True)
    _validate_in(data.get("status"), PO_STATUSES, "status")
    _validate_in(data.get("order_type"), PO_ORDER_TYPES, "order_type")
    _validate_in(data.get("receipt_status"), RECEIPT_STATUSES, "receipt_status")
    _validate_in(data.get("payment_status"), PAYMENT_STATUSES, "payment_status")

    if "total_amount" in data or "currency" in data:
        amount = data.get("total_amount", obj.total_amount)
        currency = data.get("currency", obj.currency)
        fx_rate, total_cny = await _resolve_fx(db, amount, currency)
        data["fx_rate_to_cny"] = fx_rate
        data["total_cny"] = total_cny

    # Snapshot the budget-relevant scope BEFORE mutating so we can tell
    # the reconciler whether the PO moved to a different pool, and what
    # currency the existing lock was tagged with (releases must use it).
    prev_scope = (
        obj.department, obj.cost_center, obj.category_id, obj.currency,
    )
    prev_currency = obj.currency
    for k, v in data.items():
        setattr(obj, k, v)
    await db.flush()

    # PR 3 (round-56 fix): keep the budget lock in sync with the PO's
    # new amount + status + scope on EVERY edit — not just cancel.
    # reconcile_budget_for_order handles release-on-cancel,
    # re-lock-on-reopen, amount delta, and scope re-match uniformly.
    scope_changed = (
        obj.department, obj.cost_center, obj.category_id, obj.currency,
    ) != prev_scope
    # SAVEPOINT — same rationale as create_order: a statement-level error
    # in the reconcile must not poison the outer session and silently drop
    # the PO edit on the router's subsequent commit.
    try:
        from app.domains.procurement.budget_service import (
            reconcile_budget_for_order,
        )
        async with db.begin_nested():
            await reconcile_budget_for_order(
                db, obj, scope_changed=scope_changed, prev_currency=prev_currency,
            )
    except Exception:  # noqa: BLE001
        logger.exception(
            "budget reconcile failed for PO %s status=%s",
            obj.order_number, obj.status,
        )

    return obj


@_active_actor_required
async def add_order_item(
    db: AsyncSession, order_id: uuid.UUID, payload: OrderItemCreate
) -> ProcurementOrderItem:
    # Parent lock serializes MAX+1 through the route's commit, including an
    # empty child set where there is no line row/gap to lock yet.
    order = await get_order(db, order_id, lock=True)
    max_line = await db.scalar(
        select(ProcurementOrderItem.line_no)
        .where(ProcurementOrderItem.order_id == str(order.id))
        .order_by(ProcurementOrderItem.line_no.desc())
        .limit(1)
        .with_for_update()
    ) or 0
    data = payload.model_dump()
    line_no = data.pop("line_no", None) or int(max_line) + 1
    line = ProcurementOrderItem(order_id=order.id, line_no=line_no, **data)
    try:
        async with db.begin_nested():
            db.add(line)
            await db.flush()
    except IntegrityError as exc:
        if not _is_serial_collision(exc, "uq_procurement_order_line"):
            raise
        raise AppException(
            code="PROCUREMENT_LINE_NUMBER_CONFLICT",
            message=f"line_no {line_no} already exists on order {order.id}",
            status_code=409,
        ) from exc
    return line


@_active_actor_required
async def update_order_item(
    db: AsyncSession, item_id: uuid.UUID, payload: OrderItemUpdate
) -> ProcurementOrderItem:
    obj = await db.get(ProcurementOrderItem, str(item_id))
    if obj is None:
        raise NotFoundError(f"Order item {item_id} not found")
    data = payload.model_dump(exclude_unset=True)
    _validate_in(data.get("receipt_status"), RECEIPT_STATUSES, "receipt_status")
    for k, v in data.items():
        setattr(obj, k, v)
    await db.flush()
    return obj


@_active_actor_required
async def delete_order_item(
    db: AsyncSession, item_id: uuid.UUID
) -> dict[str, Any]:
    """Hard-delete a PO line and return a full content snapshot.

    Order items have no version table (unlike contracts), so the snapshot
    is the only record of a deleted line's content — including any receipt
    history (received_at / received_qty). The caller writes it into the
    audit ``details`` so a deleted line is always reconstructable.
    """
    obj = await db.get(ProcurementOrderItem, str(item_id))
    if obj is None:
        raise NotFoundError(f"Order item {item_id} not found")
    snapshot = {
        "id": str(obj.id),
        "order_id": str(obj.order_id),
        "line_no": obj.line_no,
        "description": obj.description,
        "quantity": str(obj.quantity) if obj.quantity is not None else None,
        "unit": obj.unit,
        "unit_price": str(obj.unit_price) if obj.unit_price is not None else None,
        "line_total": str(obj.line_total) if obj.line_total is not None else None,
        "currency": obj.currency,
        "target_asset_id": str(obj.target_asset_id) if obj.target_asset_id else None,
        "target_software_id": (
            str(obj.target_software_id) if obj.target_software_id else None
        ),
        "target_license_plan_id": (
            str(obj.target_license_plan_id) if obj.target_license_plan_id else None
        ),
        "target_descriptor": obj.target_descriptor,
        "receipt_status": obj.receipt_status,
        "received_at": obj.received_at.isoformat() if obj.received_at else None,
        "received_qty": (
            str(obj.received_qty) if obj.received_qty is not None else None
        ),
        "notes": obj.notes,
    }
    await db.delete(obj)
    return snapshot


# ──────────────────────────────────────────────────────────────────────
# Payments
# ──────────────────────────────────────────────────────────────────────


async def list_payments(
    db: AsyncSession,
    *,
    keyword: str | None = None,
    supplier_id: uuid.UUID | None = None,
    order_id: uuid.UUID | None = None,
    software_catalog_id: uuid.UUID | None = None,
    status: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[ProcurementPayment], int]:
    base = select(ProcurementPayment)
    if keyword:
        like = f"%{keyword}%"
        base = base.where(
            or_(
                ProcurementPayment.payment_number.like(like),
                ProcurementPayment.invoice_number.like(like),
                ProcurementPayment.external_voucher_no.like(like),
            )
        )
    if supplier_id is not None:
        base = base.where(ProcurementPayment.supplier_id == str(supplier_id))
    if order_id is not None:
        base = base.where(ProcurementPayment.order_id == str(order_id))
    if software_catalog_id is not None:
        base = base.where(
            ProcurementPayment.software_catalog_id == str(software_catalog_id)
        )
    if status:
        _validate_in(status, PAYMENT_STATUSES, "status")
        base = base.where(ProcurementPayment.status == status)
    total = await db.scalar(select(func.count()).select_from(base.subquery()))
    rows = await db.execute(
        base.order_by(ProcurementPayment.created_at.desc())
        .limit(limit).offset(offset)
    )
    return list(rows.scalars()), int(total or 0)


async def get_payment(db: AsyncSession, payment_id: uuid.UUID) -> ProcurementPayment:
    obj = await db.get(ProcurementPayment, str(payment_id))
    if obj is None:
        raise NotFoundError(f"Payment {payment_id} not found")
    return obj


@_active_actor_required
async def create_payment(
    db: AsyncSession,
    payload: PaymentCreate,
    *,
    before_commit: Callable[[ProcurementPayment], Awaitable[None]] | None = None,
) -> ProcurementPayment:
    payload = _prepare_payment_create(payload)

    # Legacy callers without a durable Idempotency-Key retain the short
    # heuristic guard. HTTP/MCP creation paths use
    # ``create_payment_idempotent`` below instead.
    return await _create_payment_with_legacy_window(
        db, payload, before_commit=before_commit,
    )


def _validate_payment_create(payload: PaymentCreate) -> None:
    _validate_in(payload.status, PAYMENT_STATUSES, "status")
    if not any([payload.order_id, payload.contract_id, payload.software_catalog_id]):
        raise ValidationError(
            "Payment must reference at least one of order_id / contract_id / "
            "software_catalog_id"
        )


def _prepare_payment_create(payload: PaymentCreate) -> PaymentCreate:
    """Validate and canonicalize one payment intent before dedup/fingerprint."""
    _validate_payment_create(payload)
    amount = _canonical_payment_amount(payload.amount)
    currency = _canonical_payment_text(
        payload.currency,
        field="currency",
        max_length=8,
        uppercase=True,
        required=True,
    )
    assert currency is not None
    invoice_number = _canonical_payment_text(
        payload.invoice_number,
        field="invoice_number",
        max_length=128,
        uppercase=True,
    )
    payment_number = _canonical_payment_text(
        payload.payment_number,
        field="payment_number",
        max_length=64,
        uppercase=True,
    )
    return payload.model_copy(update={
        "amount": amount,
        "currency": currency,
        "invoice_number": invoice_number,
        "payment_number": payment_number,
    })


async def _create_payment_with_legacy_window(
    db: AsyncSession,
    payload: PaymentCreate,
    *,
    before_commit: Callable[[ProcurementPayment], Awaitable[None]] | None = None,
) -> ProcurementPayment:

    # Idempotency / anti-double-submit guard, mirroring the canonical
    # dedup pattern in ``apikey/service.py``. ``payment_number`` is
    # server-generated (MAX+1) on each call, so a retry never collides on
    # the unique key and would BOOK A SECOND PAYMENT — a financial row.
    # Lock-key covers every reference plus amount/currency/status/invoice;
    # the pre-check returns any existing payment with the same dimensions
    # created inside the short dedup window,
    # so a retry resolves to the original payment instead of inserting.
    from app.core.distributed_lock import (  # noqa: PLC0415
        LockBusy,
        distributed_lock,
    )

    # ``_prepare_payment_create`` canonicalizes ONCE before this helper. The
    # same model is used by Redis identity, recent-row lookup and persistence.
    amount = payload.amount
    currency = payload.currency
    # Include invoice_number in the dedup identity: two genuinely-distinct
    # payments against the same PO for the same amount/currency (e.g. two
    # equal milestone installments, or two identical monthly SaaS charges)
    # differ only by invoice_number. Without it the 60s window silently
    # dropped the second REAL payment. A true double-submit carries the same
    # invoice_number (or both empty), so the double-submit guard still holds.
    invoice_number = payload.invoice_number
    payment_number = payload.payment_number
    # Hash a structured identity rather than delimiter-joining caller strings.
    # ``currency='USD:X', invoice='Y'`` and ``currency='USD', invoice='X:Y'``
    # used to collide on the same colon-joined key and falsely reject one
    # legitimate payment. The digest also keeps invoice identifiers out of
    # Redis key listings/metrics.
    lock_key = _payment_create_lock_key(
        payload,
        amount=amount,
        currency=currency,
        invoice_number=invoice_number,
        payment_number=payment_number,
    )

    async def _check_recent_dup() -> ProcurementPayment | None:
        cutoff = datetime.now(UTC) - timedelta(seconds=_PAYMENT_DEDUP_WINDOW_SECONDS)
        stmt = select(ProcurementPayment).where(
            ProcurementPayment.amount == amount,
            ProcurementPayment.currency == currency,
            ProcurementPayment.status == payload.status,
            ProcurementPayment.created_at >= cutoff,
        )
        if invoice_number is not None:
            stmt = stmt.where(
                ProcurementPayment.invoice_number == invoice_number
            )
        else:
            # No invoice → keep the double-submit guard; tolerate a stored
            # "" or NULL as "no invoice".
            stmt = stmt.where(
                or_(
                    ProcurementPayment.invoice_number.is_(None),
                    ProcurementPayment.invoice_number == "",
                )
            )
        if payment_number is not None:
            # A distinct explicit payment number is authoritative evidence of
            # a distinct payment. Automatic numbers remain outside the query
            # because a retry has no generated number to compare yet.
            stmt = stmt.where(
                ProcurementPayment.payment_number == payment_number,
            )
        if payload.order_id is not None:
            stmt = stmt.where(
                ProcurementPayment.order_id == str(payload.order_id)
            )
        else:
            stmt = stmt.where(ProcurementPayment.order_id.is_(None))
        if payload.contract_id is not None:
            stmt = stmt.where(
                ProcurementPayment.contract_id == str(payload.contract_id)
            )
        else:
            stmt = stmt.where(ProcurementPayment.contract_id.is_(None))
        if payload.software_catalog_id is not None:
            stmt = stmt.where(
                ProcurementPayment.software_catalog_id
                == str(payload.software_catalog_id)
            )
        else:
            stmt = stmt.where(ProcurementPayment.software_catalog_id.is_(None))
        stmt = stmt.order_by(ProcurementPayment.created_at.desc())
        return (await db.execute(stmt)).scalars().first()

    try:
        # Auto-renew closes the rare but dangerous case where a DB lock wait
        # outlives the nominal 120s lease and a second writer enters before
        # payment+audit commit. Lock loss cancels this holder and converges via
        # the same 409/retry path.
        async with distributed_lock(lock_key, ttl=120, auto_renew=True):
            try:
                dup = await _check_recent_dup()
                if dup is not None:
                    return dup
                # The shared counter-first allocator owns automatic and
                # explicit payment-number serialization. Keep this outer
                # intent lock through the single payment+audit commit; no
                # second Redis serial allocator is needed or allowed.
                payment = await _create_payment_inner(db, payload)
                if before_commit is not None:
                    await before_commit(payment)
                await db.commit()
                return payment
            except Exception:
                # Roll back before releasing the lock so a retry never races
                # an aborted payment, and keep the caller's session usable.
                await db.rollback()
                raise
    except LockBusy as exc:
        # The actor lifecycle gate locking-current-reads the person row before
        # Redis acquisition. Release that DB lock and its REPEATABLE READ
        # snapshot before returning 409, so the same session can observe and
        # replay the winning payment immediately.
        await db.rollback()
        raise AppException(
            code="PAYMENT_CREATE_IN_FLIGHT",
            message=(
                "An identical payment is being processed. Please retry in a "
                "few seconds."
            ),
            status_code=409,
        ) from exc


async def _create_payment_inner(
    db: AsyncSession, payload: PaymentCreate
) -> ProcurementPayment:
    data = payload.model_dump()
    number = data.pop("payment_number") or None
    fx_rate, amount_cny = await _resolve_fx(
        db, data.get("amount"), data.get("currency")
    )
    return await _insert_numbered(
        db,
        prefix="PAY",
        model=ProcurementPayment,
        column=ProcurementPayment.payment_number,
        serial_column="payment_number",
        explicit_number=number,
        build=lambda allocated: ProcurementPayment(
            payment_number=allocated,
            fx_rate_to_cny=fx_rate,
            amount_cny=amount_cny,
            **data,
        ),
    )


async def create_supplier_idempotent(
    db: AsyncSession,
    payload: SupplierCreate,
    *,
    actor_id: uuid.UUID,
    actor: AuthUser | None,
    idempotency_key: str | None,
) -> tuple[ProcurementSupplier, bool]:
    return await _idempotent_create(
        db,
        operation="supplier",
        actor_id=actor_id,
        actor=actor,
        idempotency_key=idempotency_key,
        payload=payload,
        model=ProcurementSupplier,
        creator=lambda: create_supplier(db, payload, actor=actor),
    )


async def create_contract_idempotent(
    db: AsyncSession,
    payload: ContractCreate,
    *,
    actor_id: uuid.UUID,
    actor: AuthUser | None,
    idempotency_key: str | None,
) -> tuple[ProcurementContract, bool]:
    return await _idempotent_create(
        db,
        operation="contract",
        actor_id=actor_id,
        actor=actor,
        idempotency_key=idempotency_key,
        payload=payload,
        model=ProcurementContract,
        creator=lambda: create_contract(db, payload, actor=actor),
    )


async def create_order_idempotent(
    db: AsyncSession,
    payload: OrderCreate,
    *,
    applicant_id: uuid.UUID,
    actor_id: uuid.UUID,
    actor: AuthUser | None,
    idempotency_key: str | None,
) -> tuple[ProcurementOrder, bool]:
    fingerprint_payload = {
        "payload": payload,
        "effective_applicant_id": str(applicant_id),
    }
    return await _idempotent_create(
        db,
        operation="order",
        actor_id=actor_id,
        actor=actor,
        idempotency_key=idempotency_key,
        payload=fingerprint_payload,
        model=ProcurementOrder,
        creator=lambda: create_order(
            db, payload, applicant_id=applicant_id, actor=actor,
        ),
    )


async def create_payment_idempotent(
    db: AsyncSession,
    payload: PaymentCreate,
    *,
    actor_id: uuid.UUID,
    actor: AuthUser | None,
    idempotency_key: str | None,
    before_commit: Callable[[ProcurementPayment], Awaitable[None]] | None = None,
) -> tuple[ProcurementPayment, bool]:
    payload = _prepare_payment_create(payload)
    if not (idempotency_key or "").strip():
        # Compatibility for internal/older clients that have not adopted the
        # durable key yet. This retains the 60-second heuristic and its
        # commit-inside-lock requirement; keyed HTTP/MCP calls below use the
        # atomic platform intent instead.
        return (
            await create_payment(
                db, payload, actor=actor, before_commit=before_commit,
            ),
            False,
        )

    # Payment is the one procurement create whose service owns commit: the
    # durable intent, counter mutation, payment row and audit evidence must
    # become visible together. This also releases IdempotencyService's row
    # lock before a concurrent same-key request can replay the result.
    try:
        payment, replayed = await _idempotent_create(
            db,
            operation="payment",
            actor_id=actor_id,
            actor=actor,
            idempotency_key=idempotency_key,
            payload=payload,
            model=ProcurementPayment,
            creator=lambda: _create_payment_inner(db, payload),
        )
        if not replayed and before_commit is not None:
            await before_commit(payment)
        await db.commit()
        return payment, replayed
    except Exception:
        await db.rollback()
        raise


@_active_actor_required
async def update_payment(
    db: AsyncSession, payment_id: uuid.UUID, payload: PaymentUpdate
) -> ProcurementPayment:
    obj = await get_payment(db, payment_id)
    data = payload.model_dump(exclude_unset=True)
    _validate_in(data.get("status"), PAYMENT_STATUSES, "status")
    if "amount" in data:
        data["amount"] = _canonical_payment_amount(data["amount"])
    if "currency" in data:
        data["currency"] = _canonical_payment_text(
            data["currency"], field="currency", max_length=8,
            uppercase=True, required=True,
        )
    if "invoice_number" in data:
        data["invoice_number"] = _canonical_payment_text(
            data["invoice_number"], field="invoice_number", max_length=128,
            uppercase=True,
        )
    if "amount" in data or "currency" in data:
        amount = data.get("amount", obj.amount)
        currency = data.get("currency", obj.currency)
        fx_rate, amount_cny = await _resolve_fx(db, amount, currency)
        data["fx_rate_to_cny"] = fx_rate
        data["amount_cny"] = amount_cny
    for k, v in data.items():
        setattr(obj, k, v)
    await db.flush()
    return obj


# ──────────────────────────────────────────────────────────────────────
# Software procurement meta sidecar
# ──────────────────────────────────────────────────────────────────────


async def get_software_meta(
    db: AsyncSession, software_id: uuid.UUID, *, create_if_missing: bool = False,
) -> SoftwareProcurementMeta:
    rows = await db.execute(
        select(SoftwareProcurementMeta)
        .where(SoftwareProcurementMeta.software_id == str(software_id))
    )
    obj = rows.scalar_one_or_none()
    if obj is None:
        # Auto-create on first access — the sidecar is supposed to
        # mirror the catalog, but admins can land here before the
        # migration backfill catches a hand-created catalog row.
        catalog_exists = await db.scalar(
            select(func.count(AssetSoftwareCatalog.id)).where(
                AssetSoftwareCatalog.id == str(software_id)
            )
        )
        if not catalog_exists:
            raise NotFoundError(f"Software {software_id} not found")
        obj = SoftwareProcurementMeta(software_id=str(software_id))
        # Persist ONLY on the write path. A GET must never commit a row —
        # ``get_db`` auto-commits on a successful request, so creating here on
        # a read let any read-only role (finance/auditor/management) write an
        # unaudited sidecar row just by viewing it. Reads get a transient
        # default instead.
        if create_if_missing:
            db.add(obj)
            await db.flush()
    return obj


@_active_actor_required
async def update_software_meta(
    db: AsyncSession,
    software_id: uuid.UUID,
    payload: SoftwareProcurementMetaUpdate,
) -> SoftwareProcurementMeta:
    obj = await get_software_meta(db, software_id, create_if_missing=True)
    data = payload.model_dump(exclude_unset=True)
    for k, v in data.items():
        setattr(obj, k, v)
    await db.flush()
    return obj


# ──────────────────────────────────────────────────────────────────────
# Budgets (PR 3)
# ──────────────────────────────────────────────────────────────────────


async def list_budgets(
    db: AsyncSession,
    *,
    year: int | None = None,
    department: str | None = None,
    owner_person_id: uuid.UUID | None = None,
    include_inactive: bool = False,
    limit: int = 200,
    offset: int = 0,
) -> tuple[list, int]:
    """Return budgets, paginated. Used by both admin /admin/procurement/
    budgets and the /me/budget self-service surface (filtered by owner)."""
    from app.domains.procurement.models import ProcurementBudget
    base = select(ProcurementBudget)
    if year is not None:
        base = base.where(ProcurementBudget.year == year)
    if department:
        base = base.where(ProcurementBudget.department == department)
    if owner_person_id is not None:
        base = base.where(ProcurementBudget.owner_person_id == str(owner_person_id))
    if not include_inactive:
        base = base.where(ProcurementBudget.is_active.is_(True))
    total = await db.scalar(select(func.count()).select_from(base.subquery()))
    rows = await db.execute(
        base.order_by(
            ProcurementBudget.year.desc(),
            ProcurementBudget.department.asc(),
        ).limit(limit).offset(offset)
    )
    return list(rows.scalars()), int(total or 0)


async def get_budget(db: AsyncSession, budget_id: uuid.UUID):
    from app.domains.procurement.models import ProcurementBudget
    obj = await db.get(ProcurementBudget, str(budget_id))
    if obj is None:
        raise NotFoundError(f"Budget {budget_id} not found")
    return obj


async def list_budget_consumptions(
    db: AsyncSession, budget_id: uuid.UUID, *, limit: int = 200,
) -> list:
    from app.domains.procurement.models import ProcurementBudgetConsumption
    rows = await db.execute(
        select(ProcurementBudgetConsumption)
        .where(ProcurementBudgetConsumption.budget_id == str(budget_id))
        .order_by(ProcurementBudgetConsumption.occurred_at.desc())
        .limit(limit)
    )
    return list(rows.scalars())


@_active_actor_required
async def create_budget(db: AsyncSession, payload: dict):
    """Admin create. ``payload`` is the validated dict (router writes
    Pydantic schema first; service is permissive)."""
    from app.domains.procurement.models import ProcurementBudget
    # The route passes the raw request dict (no Pydantic pre-validation),
    # so a non-numeric total_amount ("abc") would raise an uncaught
    # InvalidOperation -> 500. Parse defensively into a clean 4xx and
    # reuse the value at insert time.
    if payload.get("total_amount") is None:
        raise ValidationError("total_amount must be > 0")
    try:
        total_amount = Decimal(str(payload["total_amount"]))
    except (InvalidOperation, ValueError) as exc:
        raise ValidationError("total_amount must be a valid number") from exc
    if total_amount <= 0:
        raise ValidationError("total_amount must be > 0")
    if not payload.get("department"):
        raise ValidationError("department is required")
    if payload.get("year") in (None, ""):
        raise ValidationError("year is required")
    try:
        year_val = int(payload["year"])
    except (TypeError, ValueError) as exc:
        raise ValidationError("year must be an integer") from exc
    cost_center = payload.get("cost_center")
    category_id = payload.get("category_id")
    currency = payload.get("currency") or "CNY"

    # Reject a second active budget for the identical scope. The DB
    # UNIQUE(year, department, cost_center, category_id, currency) is
    # *loose* on MySQL/TiDB whenever any column is NULL (NULLs compare
    # distinct), so without this guard two "any cost-center / any
    # category" pools for the same dept+year+currency could both exist —
    # then PO matching is ambiguous and the pool reads are split.
    dup_q = select(ProcurementBudget.id).where(
        ProcurementBudget.year == year_val,
        ProcurementBudget.department == payload["department"],
        ProcurementBudget.currency == currency,
        ProcurementBudget.is_active.is_(True),
        ProcurementBudget.cost_center == cost_center
        if cost_center is not None
        else ProcurementBudget.cost_center.is_(None),
        ProcurementBudget.category_id == str(category_id)
        if category_id is not None
        else ProcurementBudget.category_id.is_(None),
    )
    if (await db.execute(dup_q)).first() is not None:
        raise ValidationError(
            "An active budget for this (year, department, cost_center, "
            "category, currency) scope already exists"
        )

    obj = ProcurementBudget(
        year=year_val,
        department=payload["department"],
        cost_center=cost_center,
        category_id=category_id,
        currency=currency,
        total_amount=total_amount,
        owner_person_id=payload.get("owner_person_id"),
        notes=payload.get("notes"),
    )
    db.add(obj)
    await db.flush()
    return obj


@_active_actor_required
async def update_budget(db: AsyncSession, budget_id: uuid.UUID, payload: dict):
    from app.domains.procurement.models import ProcurementBudget
    obj = await get_budget(db, budget_id)

    # Compute the EFFECTIVE post-update scope from payload-or-current.
    # Validate BEFORE applying any setattr so a rejected edit never
    # mutates the in-memory ORM object (autoflush would otherwise push
    # the change into the open transaction before the raise).
    def _eff(key):
        return payload[key] if key in payload else getattr(obj, key)

    if "total_amount" in payload and payload["total_amount"] is not None:
        try:
            new_total = Decimal(str(payload["total_amount"]))
        except (InvalidOperation, ValueError) as exc:
            raise ValidationError(
                "total_amount must be a valid number",
            ) from exc
        if new_total <= 0:
            raise ValidationError("total_amount must be > 0")
    else:
        new_total = None

    eff_year = obj.year
    if "year" in payload:
        try:
            eff_year = int(payload["year"])
        except (TypeError, ValueError) as exc:
            raise ValidationError("year must be an integer") from exc
    eff_is_active = bool(_eff("is_active"))

    # Round-61 fix: mirror ``create_budget``'s anti-ambiguity guard.
    # PATCH /budgets/{id} accepts a raw dict, so an admin could edit
    # this budget's scope (e.g. set cost_center / category_id to NULL)
    # into a duplicate of another ACTIVE budget's effective scope. The
    # DB UNIQUE(year, department, cost_center, category_id, currency)
    # is loose on MySQL/TiDB (NULLs compare distinct), so the constraint
    # would NOT reject it — two active pools with identical scope would
    # coexist and ``find_matching_budget`` (order_by created_at desc,
    # limit 1) would silently pick one and ignore the other's ceiling,
    # splitting PO locks / pool-remaining reads. Only active rows can
    # create that ambiguity (find_matching_budget filters is_active),
    # so skip the check for an inactive result.
    if eff_is_active:
        eff_cost_center = _eff("cost_center")
        eff_category_id = _eff("category_id")
        dup_q = select(ProcurementBudget.id).where(
            ProcurementBudget.id != str(obj.id),
            ProcurementBudget.year == eff_year,
            ProcurementBudget.department == _eff("department"),
            ProcurementBudget.currency == (_eff("currency") or "CNY"),
            ProcurementBudget.is_active.is_(True),
            ProcurementBudget.cost_center == eff_cost_center
            if eff_cost_center is not None
            else ProcurementBudget.cost_center.is_(None),
            ProcurementBudget.category_id == str(eff_category_id)
            if eff_category_id is not None
            else ProcurementBudget.category_id.is_(None),
        )
        if (await db.execute(dup_q)).first() is not None:
            raise ValidationError(
                "An active budget for this (year, department, cost_center, "
                "category, currency) scope already exists"
            )

    for k in (
        "year", "department", "cost_center", "category_id",
        "currency", "owner_person_id", "notes", "is_active",
    ):
        if k in payload:
            setattr(obj, k, payload[k])
    obj.year = eff_year
    if new_total is not None:
        obj.total_amount = new_total
    await db.flush()
    return obj