"""Data reconciliation center — deep forensic of three drift classes
that PR-A's health console only surfaces as numbers.

Three lenses:

  * **Seat reconciliation** — per-plan ``vendor_license_holders >
    software_license_plans.max_seats`` (the under-licensed signal PR
    #610 added) PLUS plans whose 6-bucket identity doesn't hold
    (``vendor != matched + external + no_email + leaver_residual +
    vendor_inactive + off_paid_site``). Either is an admin-must-act
    drift state.

  * **Ownership FK integrity** — per resource type
    (scope / dns / cdn / domain / api_key / asset), rows whose
    owner_person_id is NULL OR points at a soft-deleted /
    employment_status='left' person. The "who owns this when nobody
    owns it" question.

  * **Leaver residual** — every ``identity_persons.employment_status
    = 'left'`` whose resource owner_person_id / assigned_to /
    person_id still points at them across the same six resource
    types. Cross-cuts PR-A's health-panel single-number leaver-
    residual count by giving per-resource drilldown.

This module is **read-only**; the row-level actions
(open compliance incident, force-revoke, etc.) reuse existing
admin endpoints on the relevant domain pages. The center is a
forensic surface, not a write surface.
"""
from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.apikey.models import ApiKeyAssignment
from app.domains.asset.models import (
    AssetItem,
    AssetLocation,
    AssetLocationAdmin,
    AssetOfficeAddress,
    AssetOfficeLocalAdmin,
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
    SoftwareSeatSyncStat,
)
from app.domains.cmdb.models import (
    SCOPE_TYPE_GCP_BILLING_ACCOUNT,
    CmdbCdnDomain,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbDnsRecord,
    CmdbDomain,
    owner_governed_scope_criteria,
)
from app.domains.procurement.models import (
    ProcurementContract,
    ProcurementOrder,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

# Resources in a terminal / decommissioned state are intentionally
# orphaned by their respective lifecycle flows — surfacing them on
# the "who owns this?" lens is noise, not signal. Mirrors the same
# filter the admin overview health console uses (see
# ``cmdb/admin_overview_service.py``:
# "Owner health — only meaningful for non-closed scopes (closed
# scopes are intentionally orphaned by the decommission flow)").
#
#   * cloud_scope: ``status='closed'`` OR ``decommissioned_at`` set —
#     account/subscription/project was closed via the decommission
#     flow; the cloud-side resource is gone or pending closure.
#   * cdn_domain: ``status='offline'`` — domain was taken offline at
#     the CDN provider; not billable, no governance need.
#   * domain (root): ``status='expired'`` — registration lapsed,
#     no longer in use.
#   * asset: terminal lifecycle states (``disposed`` /
#     ``disposal_pending`` / ``retired``) — physical asset is gone.
#   * api_key: already filtered to ``status='active'`` below.
#   * dns_record: ``status='disable'`` is a reversible toggle, NOT
#     terminal — the record still exists at the provider and an
#     owner is still meaningful, so DNS is left as-is.
_TERMINAL_ASSET_STATUSES: frozenset[str] = frozenset(
    {"disposed", "disposal_pending"},
)
_TERMINAL_CDN_STATUS = "offline"
_TERMINAL_DOMAIN_STATUS = "expired"
# Procurement contracts/orders in a terminal state no longer need an
# active responsible owner — a leaver owning a ``terminated`` / ``expired``
# contract or a ``completed`` / ``cancelled`` / ``rejected`` PO is historical
# attribution, not live residue. In-flight ones (draft/signed/active
# contract; draft/submitted/approved/received PO) DO need transfer.
_TERMINAL_CONTRACT_STATUSES: frozenset[str] = frozenset({"terminated", "expired"})
_TERMINAL_PO_STATUSES: frozenset[str] = frozenset(
    {"completed", "cancelled", "rejected"},
)


# ── Seat reconciliation ─────────────────────────────────────────


async def seat_reconciliation(db: AsyncSession) -> dict[str, Any]:
    """Per-plan seat-drift detail.

    Two row classes:
      * ``under_licensed`` — vendor_license_holders > plan.max_seats
      * ``identity_broken`` — sum of 6 holder buckets != vendor_license_holders

    A plan can be in both classes simultaneously.
    """
    rows = (
        await db.execute(
            select(
                SoftwareSeatSyncStat,
                SoftwareLicensePlan,
            )
            .join(
                SoftwareLicensePlan,
                SoftwareLicensePlan.id == SoftwareSeatSyncStat.plan_id,
            )
        )
    ).all()

    under: list[dict[str, Any]] = []
    broken: list[dict[str, Any]] = []
    for stat, plan in rows:
        vlh = int(stat.vendor_license_holders or 0)
        bucket_sum = (
            int(stat.holders_matched or 0)
            + int(stat.holders_external or 0)
            + int(stat.holders_no_email or 0)
            + int(stat.holders_leaver_residual or 0)
            + int(stat.holders_vendor_inactive or 0)
            + int(stat.holders_off_paid_site or 0)
        )
        common = {
            "plan_id": str(plan.id),
            "plan_name": plan.name,
            "software_id": str(plan.software_id),
            "max_seats": plan.max_seats,
            "vendor_license_holders": vlh,
            "buckets": {
                "matched": int(stat.holders_matched or 0),
                "external": int(stat.holders_external or 0),
                "no_email": int(stat.holders_no_email or 0),
                "leaver_residual": int(stat.holders_leaver_residual or 0),
                "vendor_inactive": int(stat.holders_vendor_inactive or 0),
                "off_paid_site": int(stat.holders_off_paid_site or 0),
            },
            "bucket_sum": bucket_sum,
            "provider": stat.provider,
            "product_key": stat.product_key,
            "synced_at": stat.synced_at.isoformat() if stat.synced_at else None,
        }
        if plan.max_seats is not None and vlh > plan.max_seats:
            row = dict(common)
            row["over_by"] = vlh - plan.max_seats
            under.append(row)
        if vlh != bucket_sum:
            row = dict(common)
            row["drift"] = vlh - bucket_sum
            broken.append(row)

    under.sort(key=lambda r: r.get("over_by", 0), reverse=True)
    broken.sort(key=lambda r: abs(r.get("drift", 0)), reverse=True)
    return {
        "under_licensed": under,
        "identity_broken": broken,
    }


# ── Ownership FK integrity ──────────────────────────────────────


async def _active_person_ids(db: AsyncSession) -> set[str]:
    rows = (
        await db.execute(
            select(IdentityPerson.id).where(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
            )
        )
    ).all()
    return {str(r[0]) for r in rows}


async def ownership_integrity(db: AsyncSession) -> dict[str, Any]:
    """Per-resource-type rows whose owner is NULL or non-active.

    Returns a dict keyed by resource type → list of {resource_id,
    label, owner_person_id, owner_state} rows. ``owner_state`` is
    one of ``null``, ``not_found``, ``inactive``, ``soft_deleted`` —
    drives the per-row Tag color on the UI.
    """
    active_ids = await _active_person_ids(db)

    # All persons (including inactive/deleted) so we can classify
    # the owner_state of a non-active reference.
    all_persons = (
        await db.execute(
            select(
                IdentityPerson.id,
                IdentityPerson.employment_status,
                IdentityPerson.deleted_at,
            )
        )
    ).all()
    person_by_id: dict[str, tuple[str, bool]] = {
        str(r[0]): (r[1] or "", r[2] is not None) for r in all_persons
    }

    def classify(owner_id: Any) -> str | None:
        if not owner_id:
            return "null"
        sid = str(owner_id)
        if sid in active_ids:
            return None  # healthy
        meta = person_by_id.get(sid)
        if meta is None:
            return "not_found"
        _, soft_deleted = meta
        if soft_deleted:
            return "soft_deleted"
        return "inactive"

    issues: dict[str, list[dict[str, Any]]] = {
        "cloud_scope": [],
        "dns_record": [],
        "cdn_domain": [],
        "domain": [],
        "api_key": [],
        "asset": [],
    }

    # Cloud scopes — exclude closed / decommissioned (intentionally
    # orphaned by the decommission flow; no owner needed). Also exclude
    # GCP billing-account scopes: they are intentionally ownerless (a
    # billing account has no owner-equivalent IAM role to mint — approval
    # routes to the cloud-admin team, see lifecycle_service), so flagging
    # them as owner-orphans would be a permanent false positive an admin
    # could never clear. Mirrors the same exclusion in the auto-owner-grant
    # backfill cron (scheduled_jobs/handlers.py).
    scopes = (
        await db.execute(
            select(CmdbCloudScope).where(
                # Shared base predicate — same one the leaver-residue lens
                # below, offboarding blockers and the manager team view
                # use. It supplies deleted_at / closed / decommissioned /
                # tenant-level-type; the two exclusions after it are this
                # lens's own and compose on top.
                #
                # 2026-09-07: the tenant-level clause arrived with the
                # shared predicate and it is not cosmetic — before it,
                # **all 7 rows this lens reported in prod were
                # ``legacy_billing_unit``**, i.e. the entire cloud_scope
                # output of the "who owns this when nobody owns it" lens
                # was noise. Migration 466 typed those rows precisely so
                # no owner governance applies to them, so an admin could
                # never clear one — the same permanent-false-positive
                # shape the GCP-billing exclusion below was added for.
                *owner_governed_scope_criteria(),
                CmdbCloudScope.scope_type != SCOPE_TYPE_GCP_BILLING_ACCOUNT,
                # Aliyun RD master (payer) is owner-exempt infra — it's a
                # scope only so master-allocated billing lines have a home
                # (raw.rd_master marker, AliyunCloudDriver.list_scopes). An
                # unassigned owner on it is not an orphan, same as a GCP
                # billing account above.
                func.json_extract(
                    CmdbCloudScope.raw_data, "$.rd_master",
                ).is_(None),
            )
        )
    ).scalars().all()
    for s in scopes:
        state = classify(s.owner_person_id)
        if state:
            issues["cloud_scope"].append({
                "resource_id": str(s.id),
                "label": s.display_name or s.external_id,
                "owner_person_id": (
                    str(s.owner_person_id) if s.owner_person_id else None
                ),
                "owner_state": state,
            })

    # DNS
    dns_rows = (
        await db.execute(
            select(CmdbDnsRecord).where(CmdbDnsRecord.deleted_at.is_(None))
        )
    ).scalars().all()
    for r in dns_rows:
        state = classify(r.owner_person_id)
        if state:
            issues["dns_record"].append({
                "resource_id": str(r.id),
                "label": f"{r.zone_name}/{r.name} ({r.record_type})",
                "owner_person_id": (
                    str(r.owner_person_id) if r.owner_person_id else None
                ),
                "owner_state": state,
            })

    # CDN — exclude offline (taken down at provider, no governance need).
    cdn_rows = (
        await db.execute(
            select(CmdbCdnDomain).where(
                CmdbCdnDomain.deleted_at.is_(None),
                CmdbCdnDomain.status != _TERMINAL_CDN_STATUS,
            )
        )
    ).scalars().all()
    for c in cdn_rows:
        state = classify(c.owner_person_id)
        if state:
            issues["cdn_domain"].append({
                "resource_id": str(c.id),
                "label": c.domain_name,
                "owner_person_id": (
                    str(c.owner_person_id) if c.owner_person_id else None
                ),
                "owner_state": state,
            })

    # Root domains — exclude expired (registration lapsed).
    domain_rows = (
        await db.execute(
            select(CmdbDomain).where(
                CmdbDomain.deleted_at.is_(None),
                CmdbDomain.status != _TERMINAL_DOMAIN_STATUS,
            )
        )
    ).scalars().all()
    for d in domain_rows:
        state = classify(d.owner_person_id)
        if state:
            issues["domain"].append({
                "resource_id": str(d.id),
                "label": d.domain_name,
                "owner_person_id": (
                    str(d.owner_person_id) if d.owner_person_id else None
                ),
                "owner_state": state,
            })

    # Api keys (active)
    apikey_rows = (
        await db.execute(
            select(ApiKeyAssignment).where(ApiKeyAssignment.status == "active")
        )
    ).scalars().all()
    for a in apikey_rows:
        state = classify(a.assigned_to)
        if state:
            issues["api_key"].append({
                "resource_id": str(a.id),
                "label": a.request_number,
                "owner_person_id": (
                    str(a.assigned_to) if a.assigned_to else None
                ),
                "owner_state": state,
            })

    # Assets (in use) — exclude terminal lifecycle states.  Disposed
    # / retired assets often retain ``assigned_to`` for audit-trail
    # purposes (PR #495 preserves the last-holder attribution on
    # disposal); without this filter every disposed laptop whose last
    # holder happens to be inactive shows up as an "inactive owner"
    # row, drowning out the actually-actionable signal.
    asset_rows = (
        await db.execute(
            select(AssetItem).where(
                AssetItem.deleted_at.is_(None),
                AssetItem.assigned_to.isnot(None),
                AssetItem.status.notin_(_TERMINAL_ASSET_STATUSES),
            )
        )
    ).scalars().all()
    for a in asset_rows:
        state = classify(a.assigned_to)
        if state:
            issues["asset"].append({
                "resource_id": str(a.id),
                "label": a.asset_tag or (a.asset_name or "asset"),
                "owner_person_id": (
                    str(a.assigned_to) if a.assigned_to else None
                ),
                "owner_state": state,
            })

    counts = {k: len(v) for k, v in issues.items()}
    return {"counts": counts, "issues": issues}


# ── Leaver residual ─────────────────────────────────────────────


async def leaver_residual(db: AsyncSession) -> dict[str, Any]:
    """Every ``employment_status='left'`` person whose resources
    still point at them.

    Differs from ``ownership_integrity`` which is broader (NULL +
    inactive + soft_deleted + not_found); this is strictly the
    "they're gone but still hold things" lens, scoped to active
    grant / in-use asset rows (i.e. the things that materially
    matter when somebody's left).
    """
    # Pull every left person once.
    left_rows = (
        await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.employment_status == "left",
                IdentityPerson.deleted_at.is_(None),
            )
        )
    ).scalars().all()
    left_ids = [str(p.id) for p in left_rows]
    # UUID-typed variant for IN-clauses against native ``Uuid`` columns
    # (cloud/asset/procurement person FKs); ``left_ids`` (str) is for the
    # ``String(36)`` ``person_id`` columns and the ``left_meta`` lookup.
    left_uuid = [p.id for p in left_rows]
    left_meta = {
        str(p.id): {
            "display_name": p.display_name,
            "work_email": p.work_email,
            "termination_date": (
                p.termination_date.isoformat() if p.termination_date else None
            ),
        }
        for p in left_rows
    }
    if not left_ids:
        return {"persons": [], "rows": []}

    rows: list[dict[str, Any]] = []

    # ── Active grants
    grants = (
        await db.execute(
            select(CmdbCloudAccessGrant).where(
                CmdbCloudAccessGrant.person_id.in_(left_ids),
                CmdbCloudAccessGrant.status == "active",
            )
        )
    ).scalars().all()
    for g in grants:
        rows.append({
            "resource_type": "cloud_access_grant",
            "resource_id": str(g.id),
            "label": f"{g.role_display_name or g.role_key}",
            "leaver_person_id": g.person_id,
            "leaver": left_meta.get(str(g.person_id), {}),
        })

    # ── Scopes still owned — exclude closed / decommissioned scopes
    # (intentionally orphaned by the decommission flow; carrying a
    # leaver's FK there is expected, not residue).
    #
    # 2026-09-07: this was the platform's most-correct copy of the
    # predicate and is now the shared one. It gained the scope_type
    # clause in the swap — a leaver's ``legacy_billing_unit`` FK is the
    # same kind of bookkeeping as a closed scope's, and reporting it as
    # residue asked an admin to reassign a row the console won't show.
    scopes = (
        await db.execute(
            select(CmdbCloudScope).where(
                CmdbCloudScope.owner_person_id.in_(left_ids),
                *owner_governed_scope_criteria(),
            )
        )
    ).scalars().all()
    for s in scopes:
        rows.append({
            "resource_type": "cloud_scope_owner",
            "resource_id": str(s.id),
            "label": s.display_name or s.external_id,
            "leaver_person_id": str(s.owner_person_id),
            "leaver": left_meta.get(str(s.owner_person_id), {}),
        })

    # ── DNS still owned
    dns = (
        await db.execute(
            select(CmdbDnsRecord).where(
                CmdbDnsRecord.owner_person_id.in_(left_ids),
                CmdbDnsRecord.deleted_at.is_(None),
            )
        )
    ).scalars().all()
    for r in dns:
        rows.append({
            "resource_type": "dns_record",
            "resource_id": str(r.id),
            "label": f"{r.zone_name}/{r.name}",
            "leaver_person_id": str(r.owner_person_id),
            "leaver": left_meta.get(str(r.owner_person_id), {}),
        })

    # ── CDN still owned — exclude offline domains.
    cdn = (
        await db.execute(
            select(CmdbCdnDomain).where(
                CmdbCdnDomain.owner_person_id.in_(left_ids),
                CmdbCdnDomain.deleted_at.is_(None),
                CmdbCdnDomain.status != _TERMINAL_CDN_STATUS,
            )
        )
    ).scalars().all()
    for c in cdn:
        rows.append({
            "resource_type": "cdn_domain",
            "resource_id": str(c.id),
            "label": c.domain_name,
            "leaver_person_id": str(c.owner_person_id),
            "leaver": left_meta.get(str(c.owner_person_id), {}),
        })

    # ── Domains still owned — exclude expired registrations.
    domains = (
        await db.execute(
            select(CmdbDomain).where(
                CmdbDomain.owner_person_id.in_(left_ids),
                CmdbDomain.deleted_at.is_(None),
                CmdbDomain.status != _TERMINAL_DOMAIN_STATUS,
            )
        )
    ).scalars().all()
    for d in domains:
        rows.append({
            "resource_type": "domain",
            "resource_id": str(d.id),
            "label": d.domain_name,
            "leaver_person_id": str(d.owner_person_id),
            "leaver": left_meta.get(str(d.owner_person_id), {}),
        })

    # ── Active api keys
    apikeys = (
        await db.execute(
            select(ApiKeyAssignment).where(
                ApiKeyAssignment.assigned_to.in_(left_uuid),
                ApiKeyAssignment.status == "active",
            )
        )
    ).scalars().all()
    for a in apikeys:
        rows.append({
            "resource_type": "api_key",
            "resource_id": str(a.id),
            "label": a.request_number,
            "leaver_person_id": str(a.assigned_to),
            "leaver": left_meta.get(str(a.assigned_to), {}),
        })

    # ── Assets in use — exclude disposed / retired (audit-trail
    # ``assigned_to`` is intentionally preserved on disposal; carrying
    # a leaver's FK there is expected provenance, not residue).
    assets = (
        await db.execute(
            select(AssetItem).where(
                AssetItem.assigned_to.in_(left_uuid),
                AssetItem.deleted_at.is_(None),
                AssetItem.status.notin_(_TERMINAL_ASSET_STATUSES),
            )
        )
    ).scalars().all()
    for a in assets:
        rows.append({
            "resource_type": "asset",
            "resource_id": str(a.id),
            "label": a.asset_tag or (a.asset_name or "asset"),
            "leaver_person_id": str(a.assigned_to),
            "leaver": left_meta.get(str(a.assigned_to), {}),
        })

    def _emit_owner_rows(
        resource_type: str,
        resource_id: str,
        label_base: str,
        roles: list[tuple[str, Any]],
    ) -> None:
        """Append one residual row per leaver-held owner role on a resource.

        A single contract/PO can name a leaver in more than one role
        (business / procurement / it / finance) — each is its own piece of
        unattributed responsibility, so each gets a row tagged with the role.
        """
        for role, owner_id in roles:
            if owner_id is None or str(owner_id) not in left_meta:
                continue
            rows.append({
                "resource_type": resource_type,
                "resource_id": resource_id,
                "label": f"{label_base} ({role})",
                "leaver_person_id": str(owner_id),
                "leaver": left_meta.get(str(owner_id), {}),
            })

    # ── Procurement contracts still owned — exclude terminal (terminated /
    # expired) contracts. A leaver named on an in-flight contract is
    # unattributed responsibility, invisible to both the offboarding scan
    # and the other lenses. All four owner/contact roles are checked.
    contracts = (
        await db.execute(
            select(ProcurementContract).where(
                or_(
                    ProcurementContract.business_owner_id.in_(left_uuid),
                    ProcurementContract.procurement_owner_id.in_(left_uuid),
                    ProcurementContract.it_contact_id.in_(left_uuid),
                    ProcurementContract.finance_contact_id.in_(left_uuid),
                ),
                ProcurementContract.status.notin_(_TERMINAL_CONTRACT_STATUSES),
            )
        )
    ).scalars().all()
    for c in contracts:
        _emit_owner_rows(
            "procurement_contract",
            str(c.id),
            c.title or c.contract_number or "contract",
            [
                ("business_owner", c.business_owner_id),
                ("procurement_owner", c.procurement_owner_id),
                ("it_contact", c.it_contact_id),
                ("finance_contact", c.finance_contact_id),
            ],
        )

    # ── Procurement orders still owned — exclude terminal POs.
    orders = (
        await db.execute(
            select(ProcurementOrder).where(
                or_(
                    ProcurementOrder.business_owner_id.in_(left_uuid),
                    ProcurementOrder.procurement_owner_id.in_(left_uuid),
                    ProcurementOrder.it_contact_id.in_(left_uuid),
                    ProcurementOrder.finance_contact_id.in_(left_uuid),
                ),
                ProcurementOrder.status.notin_(_TERMINAL_PO_STATUSES),
            )
        )
    ).scalars().all()
    for o in orders:
        # Same four live-responsibility roles as the contract block — a
        # leaver named only as the IT / finance contact of an in-flight PO
        # is residue too. ``applicant_id`` (the original requester) is NOT
        # surfaced: it's a mandatory historical field on every PO with no
        # contract analog, so it would flood the lens with non-ownership rows.
        _emit_owner_rows(
            "procurement_order",
            str(o.id),
            o.order_number or "PO",
            [
                ("business_owner", o.business_owner_id),
                ("procurement_owner", o.procurement_owner_id),
                ("it_contact", o.it_contact_id),
                ("finance_contact", o.finance_contact_id),
            ],
        )

    # ── Asset-location administrator still held by a leaver. The
    # ``asset_admin`` role is fully auto-derived from this table, so a
    # departed administrator means the location's coverage silently
    # rests on someone who has left until an operator reassigns it.
    # Exclude deactivated locations — a retired location's admin is not
    # live residue, matching the terminal-state exclusion every other
    # block applies.
    location_admins = (
        await db.execute(
            select(AssetLocationAdmin, AssetLocation)
            .join(
                AssetLocation,
                AssetLocation.id == AssetLocationAdmin.location_id,
            )
            .where(
                AssetLocationAdmin.person_id.in_(left_uuid),
                AssetLocation.is_active.is_(True),
            )
        )
    ).all()
    for admin_row, location in location_admins:
        rows.append({
            "resource_type": "location_admin",
            "resource_id": f"{location.id}:{admin_row.person_id}",
            "label": location.name,
            "leaver_person_id": str(admin_row.person_id),
            "leaver": left_meta.get(str(admin_row.person_id), {}),
        })

    # ── 当地 Admin still held by a leaver. Grants no authority, but a
    # departed local contact is who a new hire in that office is told
    # to find on day one, so it is real residue for the HR flows.
    local_admins = (
        await db.execute(
            select(AssetOfficeLocalAdmin, AssetOfficeAddress)
            .join(
                AssetOfficeAddress,
                AssetOfficeAddress.id == AssetOfficeLocalAdmin.office_id,
            )
            .where(
                AssetOfficeLocalAdmin.person_id.in_(left_uuid),
                AssetOfficeAddress.is_active.is_(True),
            )
        )
    ).all()
    for admin_row, office in local_admins:
        rows.append({
            "resource_type": "office_local_admin",
            "resource_id": f"{office.id}:{admin_row.person_id}",
            "label": office.office_name or office.city,
            "leaver_person_id": str(admin_row.person_id),
            "leaver": left_meta.get(str(admin_row.person_id), {}),
        })

    # Per-resource counts
    counts: dict[str, int] = {}
    persons_with_residue: set[str] = set()
    for r in rows:
        counts[r["resource_type"]] = counts.get(r["resource_type"], 0) + 1
        persons_with_residue.add(r["leaver_person_id"])

    return {
        "counts": counts,
        "persons_count": len(persons_with_residue),
        "rows": rows,
    }


# ── Top-level aggregator (one round-trip overview) ──────────────


async def cost_drift(db: AsyncSession) -> dict[str, Any]:
    """Cost lens — rollup↔live-recompute drift for recent closed months.

    Thin pass-through to the ai_spending domain's
    :func:`app.domains.ai_spending.cost_drift.cost_rollup_drift`. The
    reconciliation center is a cross-domain forensic reader (it already
    joins asset / cmdb / procurement / apikey models), and ai_spending is
    a reporting/coordinator-class service per PLATFORM.md's domain-rule
    carve-out, so calling its public reconciliation function from here is
    sanctioned. Imported lazily so the heavy ai_spending service graph
    isn't pulled in for callers that only need the seat/ownership/leaver
    lenses.
    """
    from app.domains.ai_spending.cost_drift import cost_rollup_drift
    return await cost_rollup_drift(db)


async def get_reconciliation_overview(db: AsyncSession) -> dict[str, Any]:
    """Single payload for the /admin/reconciliation page.

    Each lens runs independently — per-lens exception is captured
    inline so one bad slice doesn't 500 the whole page.
    """
    out: dict[str, Any] = {}
    for key, builder in (
        ("seat", seat_reconciliation(db)),
        ("ownership", ownership_integrity(db)),
        ("leaver", leaver_residual(db)),
        ("cost", cost_drift(db)),
        ("provisioning_gap", provisioning_gap(db)),
    ):
        try:
            out[key] = await builder
        except Exception as exc:  # noqa: BLE001
            out[key] = {"status": "error", "error": repr(exc)}
            # A DB/operational error (statement timeout on a large scan,
            # deadlock) leaves the shared session in a rollback-required
            # state. Without an explicit rollback the NEXT lens's first
            # query raises PendingRollbackError, cascading one bad slice
            # into "all lenses error". Roll back so the isolation this
            # loop advertises actually holds for DB errors, not just
            # pure-Python ones. Best-effort — a failed rollback (rare)
            # is swallowed so we still return the lenses gathered so far.
            try:
                await db.rollback()
            except Exception:  # noqa: BLE001
                logger.exception(
                    "reconciliation overview: rollback after '%s' lens "
                    "error also failed", key,
                )
    return out


# silence unused-import lint until ownership predicate uses them
_unused = (and_, or_, func)


# ── Provisioning gap (2026-08-06) ──────────────────────────────────
#
# The lens that would have caught the Zoom-seat incident on day one.
#
# On 2026-08-06 an approved Feishu seat request never reached Zoom: an FK
# violation on the queued provisioning task poisoned the transaction, the
# approval callback rolled back, and the row sat ``pending`` for 20 hours.
# Nothing surfaced it — the retry cron reported ``last_status=ok`` because
# its health gate only fired when EVERY row failed, and "active in Helix,
# never granted at the vendor" has no representation in any UI. It took a
# human comparing the Feishu card against the Zoom admin console.
#
# Signal design matters more than coverage here. Two naive predicates were
# measured against production and rejected:
#
#   * "active seat with no completed grant task" → 2,700+ rows. Most seats
#     are IMPORTED by the vendor sync crons and never had a Helix task, so
#     the absence of one means nothing.
#   * "revoke task still pending" → 39 rows, ALL of them deliberately
#     deferred to the leaver's last working day via ``not_before``.
#
# Both would have made this tab a wall of false positives, which is the
# same as having no tab. The predicates below measured 0 / 2 / 0 / 0 on
# production after the fix — low enough that a non-zero count is worth
# acting on.

#: A task is only late once its defer gate has passed. Offboarding schedules
#: revokes for the leaver's last working day, so ``not_before`` in the future
#: is correct-and-waiting, not stuck. The dispatcher runs every 30s.
_TASK_OVERDUE_AFTER_HOURS = 2

#: Vendor states meaning "this instance will never move again". A Helix row
#: still ``pending`` against one of these is the incident's exact signature.
_TERMINAL_VENDOR_STATES = (
    "APPROVED", "REJECTED", "CANCELED", "CANCELLED", "DELETED",
)


async def provisioning_gap(db: AsyncSession) -> dict[str, Any]:
    """Access Helix believes it granted but never actually delivered.

    Four independent buckets — a non-empty one is actionable on its own:

    ``overdue_tasks``
        Queued vendor calls whose defer gate has passed and which are still
        not done.
    ``failed_tasks``
        Vendor calls that gave up. Terminal, and invisible outside the
        provisioning admin page today.
    ``unprovisioned_seats``
        Seats Helix ORIGINATED (an approval stamped them) that are ACTIVE
        while no grant task ever completed — "we told the employee they
        have it; the vendor was never told". Vendor-imported seats are
        excluded (they legitimately have no task), and so are ``pending``
        seats: one is not yet granted by definition, and flagging a seat
        whose approval is still legitimately outstanding at Feishu was a
        measured false positive. That stuck-forever case belongs to
        ``stale_approvals``, so the two buckets stay disjoint.
    ``stale_approvals``
        Rows still ``pending`` in Helix whose last-seen vendor snapshot is
        already terminal. Read from the stored
        ``source_payload.latest_instance_detail`` the drift reconciler
        refreshes every 15 minutes, so it costs no Feishu round-trip — and
        would have lit up within 15 minutes of the incident.
    """
    from datetime import UTC, datetime, timedelta

    from app.approval.models import ApprovalRequest
    from app.domains.provisioning.models import (
        STATUS_FAILED,
        STATUS_IN_PROGRESS,
        STATUS_PENDING,
        ProvisioningTask,
        SoftwareConnector,
    )

    now = datetime.now(UTC).replace(tzinfo=None)
    overdue_cutoff = now - timedelta(hours=_TASK_OVERDUE_AFTER_HOURS)

    def _task_select():
        return (
            select(ProvisioningTask, AssetSoftwareCatalog, IdentityPerson)
            .outerjoin(
                AssetSoftwareCatalog,
                AssetSoftwareCatalog.id == ProvisioningTask.software_id,
            )
            .outerjoin(
                IdentityPerson, IdentityPerson.id == ProvisioningTask.user_id,
            )
        )

    overdue_rows = (
        await db.execute(
            _task_select()
            .where(
                ProvisioningTask.status.in_((STATUS_PENDING, STATUS_IN_PROGRESS)),
                or_(
                    ProvisioningTask.not_before.is_(None),
                    ProvisioningTask.not_before <= now,
                ),
                ProvisioningTask.created_at < overdue_cutoff,
            )
            .order_by(ProvisioningTask.created_at)
            .limit(200)
        )
    ).all()

    failed_rows = (
        await db.execute(
            _task_select()
            .where(ProvisioningTask.status == STATUS_FAILED)
            .order_by(ProvisioningTask.created_at.desc())
            .limit(200)
        )
    ).all()

    def _task_row(task: Any, software: Any, person: Any) -> dict[str, Any]:
        return {
            "task_id": str(task.id),
            "action": task.action,
            "target_type": task.target_type,
            "status": task.status,
            "execution_mode": task.execution_mode,
            "attempt_count": task.attempt_count,
            "software_name": software.name if software else None,
            "user_email": person.work_email if person else None,
            "user_display_name": person.display_name if person else None,
            "created_at": task.created_at.isoformat() if task.created_at else None,
            "not_before": task.not_before.isoformat() if task.not_before else None,
            "age_hours": (
                round((now - task.created_at).total_seconds() / 3600, 1)
                if task.created_at else None
            ),
        }

    overdue = [_task_row(t, s, p) for t, s, p in overdue_rows]
    failed = [_task_row(t, s, p) for t, s, p in failed_rows]

    # Helix-originated seats whose vendor grant never landed. The
    # ``approval_request_id`` stamp is what marks a seat as request-driven
    # (NB: despite the name it holds an employee_requests id — see
    # PLATFORM.md §17). Restricted to software that actually HAS a
    # connector, since manual-provisioning software never queues a task.
    seat_rows = (
        await db.execute(
            select(
                SoftwareSeatAllocation,
                SoftwareLicensePlan,
                AssetSoftwareCatalog,
                IdentityPerson,
            )
            .join(
                SoftwareLicensePlan,
                SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id,
            )
            .join(
                AssetSoftwareCatalog,
                AssetSoftwareCatalog.id == SoftwareLicensePlan.software_id,
            )
            .outerjoin(
                IdentityPerson,
                IdentityPerson.id == SoftwareSeatAllocation.user_id,
            )
            .where(
                SoftwareSeatAllocation.approval_request_id.is_not(None),
                SoftwareSeatAllocation.status == "active",
                select(SoftwareConnector.software_id)
                .where(
                    SoftwareConnector.software_id == AssetSoftwareCatalog.id,
                    SoftwareConnector.is_active.is_(True),
                )
                .exists(),
                ~select(ProvisioningTask.id)
                .where(
                    ProvisioningTask.target_id == SoftwareSeatAllocation.id,
                    ProvisioningTask.action == "grant",
                    ProvisioningTask.status == "completed",
                )
                .exists(),
            )
            .order_by(SoftwareSeatAllocation.created_at.desc())
            .limit(200)
        )
    ).all()

    unprovisioned = [
        {
            "seat_id": str(seat.id),
            "seat_status": seat.status,
            "software_name": software.name,
            "plan_name": plan.name,
            "user_email": person.work_email if person else None,
            "user_display_name": person.display_name if person else None,
            "allocated_at": (
                seat.allocated_at.isoformat() if seat.allocated_at else None
            ),
        }
        for seat, plan, software, person in seat_rows
    ]

    vendor_status = func.json_unquote(
        func.json_extract(
            ApprovalRequest.source_payload,
            "$.latest_instance_detail.status",
        )
    )
    stale_rows = (
        await db.execute(
            select(
                ApprovalRequest,
                IdentityPerson,
                vendor_status.label("vendor_status"),
            )
            .outerjoin(
                IdentityPerson,
                IdentityPerson.id == ApprovalRequest.applicant_id,
            )
            .where(
                ApprovalRequest.status == "pending",
                vendor_status.in_(_TERMINAL_VENDOR_STATES),
            )
            .order_by(ApprovalRequest.created_at)
            .limit(200)
        )
    ).all()

    stale_approvals = [
        {
            "approval_request_id": str(ar.id),
            "approval_type": ar.approval_type,
            "external_id": ar.external_id,
            "helix_status": ar.status,
            "vendor_status": vs,
            "applicant_email": person.work_email if person else None,
            "created_at": ar.created_at.isoformat() if ar.created_at else None,
            "updated_at": ar.updated_at.isoformat() if ar.updated_at else None,
        }
        for ar, person, vs in stale_rows
    ]

    counts = {
        "overdue_tasks": len(overdue),
        "failed_tasks": len(failed),
        "unprovisioned_seats": len(unprovisioned),
        "stale_approvals": len(stale_approvals),
    }
    return {
        "status": "attention" if any(counts.values()) else "ok",
        "counts": counts,
        "overdue_after_hours": _TASK_OVERDUE_AFTER_HOURS,
        "overdue_tasks": overdue,
        "failed_tasks": failed,
        "unprovisioned_seats": unprovisioned,
        "stale_approvals": stale_approvals,
    }