"""离职账号「终态保障」—— 从「我撤了权限」到「他真的进不去了」。

这个模块只回答一个问题,并且把答案落库:**每一个离职的人,在每一个身份
系统里,现在还能不能登录。**

它存在的理由是一次具体的故障。Helix 的 Jira 撤权一直工作正常,而把离职者
账号真正置为 inactive 的是组织侧的 IdP 目录同步 —— 一条 Helix 依赖、却既
不拥有也不观察的链路。2026-08-13 那条同步被一次设置页保存关掉,连续 16 天
没有任何人知道:每个离职者的 ``product_access`` 都是空的(所以每日对账天天
报「干净」),而他们的账号全都活着。**看的是一个字段,出事的是另一个字段。**

所以这里的取数原则只有一条:**判据必须来自厂商此刻怎么说,而不是 Helix
以为自己做过什么。** 一次成功的 HTTP 200 不是证据,下一轮把名册拉回来
对一遍才是。

三种判定,刻意分开而不是合成一个数:

``ok``       厂商侧账号不存在、或存在但不能登录 —— 这个人确实出不去了。
``at_risk``  账号还能登录,而且没有任何人为此负过责。这是要人去处理的那一格。
``exempt``   有一条明确的保留决定盖着 —— ``external_account_registry`` 里的
             已知分类,而这些分类**本身就是决定**:``external_guest`` 在模型
             里的定义就是**「CBG保留」**,``shared_service`` 是公共服务账号,
             ``contractor`` 是外包。

             **不要把它当成缺口。** 这一格最初写成了「豁免不等于没事」,按
             note / owner / created_by 是否为空额外算一个「无主豁免」并标红。
             三个输入全都判错了:
               · ``owner_person_id`` 在生产上 91 行里 91 行是 NULL —— 没有
                 任何代码写它,所以「没有 owner」不携带任何信息;
               · ``created_by`` 为空只表示**系统写的**,而对 CBG 保留来说
                 系统写正是**正常路径**:离职流程做一次实时的飞书 CBG 部门
                 判定,命中就调 ``ensure_retained_classification`` 落一行;
               · 而那一行本来就是 2026-07 为了**让离职对账不要每天告警一个
                 有据可查的保留**才加的 —— 红色「无主豁免」KPI 等于把那个
                 噪音原封不动又造了一遍。
             所以这里只报**决定是什么**(分类,以及有谁 / 什么时候,如果
             有),不对它做质量评判。
``unknown``  这一轮没读到厂商。永远不折叠进 ``ok``。
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import Any
from uuid import UUID

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

from app.domains.offboarding.models import (
    ASSURANCE_SYSTEM_ATLASSIAN,
    ASSURANCE_SYSTEM_GOOGLE,
    ASSURANCE_SYSTEM_IDENTITY_SYNC_PREFIX,
    HEARTBEAT_OK,
    HEARTBEAT_STALE,
    HEARTBEAT_UNKNOWN,
    VERDICT_AT_RISK,
    VERDICT_EXEMPT,
    VERDICT_OK,
    VERDICT_UNKNOWN,
    VSTATE_ABSENT,
    VSTATE_ACTIVE,
    VSTATE_DISABLED,
    VSTATE_UNKNOWN,
    IdentityChainHeartbeat,
    OffboardingAccountAssurance,
    identity_sync_assurance_system,
)

logger = logging.getLogger(__name__)

# Chain keys — stable identifiers for the dependencies Helix watches but
# does not own.
CHAIN_ATLASSIAN_IDP_SYNC = "atlassian.google_directory_sync"

# An exemption carrying none of {owner, note, created_by} is not a
# decision anybody can be asked about. Surfaced separately so "someone
# decided to keep this open" and "a bulk import touched this row once"
# stop looking identical on the page.
# What each KNOWN classification MEANS, so the page can say 「CBG保留」
# instead of echoing a database enum at the reader. Values come from
# ``asset.models.EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS``.
RETENTION_KIND_KEYS = {
    "external_guest": "cbg",
    "shared_service": "shared_service",
    "contractor": "contractor",
}


def _now() -> datetime:
    return datetime.now(UTC)


async def registry_exemptions(
    db: AsyncSession, emails: list[str],
) -> dict[str, dict[str, Any]]:
    """Map lower-cased email → the retention decision covering it.

    Mirrors ``OffboardingService._is_google_account_retained``'s
    predicate (a row whose ``classification`` is one of the KNOWN ones)
    so the page cannot disagree with the enforcement path about who is
    exempt — two places computing "is this person exempt" differently is
    how a governance surface starts lying.
    """
    if not emails:
        return {}
    from app.domains.asset.models import (
        EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS,
        ExternalAccountRegistry,
    )

    wanted = {(e or "").strip().lower() for e in emails if e}
    rows = (
        await db.execute(
            select(ExternalAccountRegistry).where(
                func.lower(ExternalAccountRegistry.email).in_(wanted),
            )
        )
    ).scalars().all()
    out: dict[str, dict[str, Any]] = {}
    for r in rows:
        if r.classification not in EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS:
            continue
        out[(r.email or "").strip().lower()] = {
            "registry_id": str(r.id),
            "classification": r.classification,
            "note": getattr(r, "note", None),
            "owner_person_id": (
                str(r.owner_person_id) if getattr(r, "owner_person_id", None)
                else None
            ),
            "created_by": (
                str(r.created_by) if getattr(r, "created_by", None) else None
            ),
            "created_at": (
                r.created_at.isoformat() if getattr(r, "created_at", None)
                else None
            ),
            # ``created_by`` NULL means the row came from offboarding's
            # automated CBG-department check, not that nobody decided.
            # Provenance, deliberately NOT a quality score.
            "system_written": not getattr(r, "created_by", None),
        }
    return out


def verdict_for(
    vendor_state: str, *, exemption: dict[str, Any] | None,
) -> tuple[str, str | None]:
    """``(verdict, exempt_reason)`` from vendor truth + retention decision.

    Order matters: a leaver who cannot sign in is ``ok`` **even when an
    exemption exists**. Reporting "exempt" for an account that is already
    closed would inflate the exemption count with rows nobody needs to
    look at, and the whole point of the exempt bucket is that every row
    in it is a live account somebody chose to keep alive.
    """
    if vendor_state in (VSTATE_ABSENT, VSTATE_DISABLED):
        return VERDICT_OK, None
    if vendor_state == VSTATE_UNKNOWN:
        return VERDICT_UNKNOWN, None
    if exemption:
        return VERDICT_EXEMPT, (
            f"registry:{exemption.get('classification')}"
        )
    return VERDICT_AT_RISK, None


async def upsert_rows(
    db: AsyncSession,
    *,
    system: str,
    rows: list[dict[str, Any]],
    evidence_at: datetime | None = None,
) -> dict[str, int]:
    """Idempotently write one sweep's worth of observations for a system.

    ``rows`` items: ``{person_id, vendor_state, vendor_account_id,
    helix_action, verdict, exempt_reason, detail}``.

    ``at_risk_since`` is preserved across runs while the row stays
    at_risk and cleared the moment it stops — that field is what lets the
    page say "open for 41 days" instead of "open", and a number of days
    is what actually makes somebody act.
    """
    stamp = evidence_at or _now()
    existing = {
        (r.person_id, r.system): r
        for r in (
            await db.execute(
                select(OffboardingAccountAssurance).where(
                    OffboardingAccountAssurance.system == system,
                )
            )
        ).scalars().all()
    }
    counts = {"created": 0, "updated": 0, "at_risk": 0, "exempt": 0, "ok": 0}
    seen: set[tuple[Any, str]] = set()
    for row in rows:
        pid = row["person_id"]
        if isinstance(pid, str):
            pid = UUID(pid)
        key = (pid, system)
        seen.add(key)
        cur = existing.get(key)
        verdict = row["verdict"]
        counts[verdict] = counts.get(verdict, 0) + 1
        if cur is None:
            cur = OffboardingAccountAssurance(
                person_id=pid, system=system,
            )
            db.add(cur)
            counts["created"] += 1
        else:
            counts["updated"] += 1
        cur.vendor_state = row["vendor_state"]
        cur.vendor_account_id = (
            str(row["vendor_account_id"])[:128]
            if row.get("vendor_account_id") else None
        )
        cur.helix_action = (
            str(row["helix_action"])[:48] if row.get("helix_action") else None
        )
        cur.exempt_reason = (
            str(row["exempt_reason"])[:64] if row.get("exempt_reason") else None
        )
        cur.detail = row.get("detail")
        cur.evidence_at = stamp
        if verdict == VERDICT_AT_RISK:
            if cur.verdict != VERDICT_AT_RISK or cur.at_risk_since is None:
                cur.at_risk_since = stamp
        else:
            cur.at_risk_since = None
        cur.verdict = verdict

    # Rows for people who are no longer in scope (rehired, hard-deleted,
    # or simply no longer returned by the sweep) must not linger claiming
    # a stale verdict — a governance page whose worst number is a fossil
    # trains people to ignore it.
    stale = [r for k, r in existing.items() if k not in seen]
    for r in stale:
        await db.delete(r)
    counts["removed"] = len(stale)
    await db.flush()
    return counts


async def record_heartbeat(
    db: AsyncSession,
    *,
    chain_key: str,
    last_beat_at: datetime | None,
    expected_period_minutes: int,
    detail: dict[str, Any] | None = None,
    read_ok: bool = True,
) -> dict[str, Any]:
    """Write one beat and classify the chain.

    ``read_ok=False`` (the vendor could not be read at all) lands
    ``unknown``, never ``ok`` — "I could not check" and "it is fine" are
    different answers and only one of them is safe to show in green.
    """
    now = _now()
    row = (
        await db.execute(
            select(IdentityChainHeartbeat).where(
                IdentityChainHeartbeat.chain_key == chain_key,
            )
        )
    ).scalars().first()
    if row is None:
        row = IdentityChainHeartbeat(chain_key=chain_key)
        db.add(row)
    row.expected_period_minutes = expected_period_minutes
    row.checked_at = now
    row.detail = detail
    if not read_ok:
        row.status = HEARTBEAT_UNKNOWN
    elif last_beat_at is None:
        row.status = HEARTBEAT_STALE
    else:
        if last_beat_at.tzinfo is None:
            last_beat_at = last_beat_at.replace(tzinfo=UTC)
        row.last_beat_at = last_beat_at
        age_min = (now - last_beat_at).total_seconds() / 60.0
        row.status = (
            HEARTBEAT_OK if age_min <= expected_period_minutes
            else HEARTBEAT_STALE
        )
    await db.flush()
    return {
        "chain_key": chain_key,
        "status": row.status,
        "last_beat_at": (
            row.last_beat_at.isoformat() if row.last_beat_at else None
        ),
        "expected_period_minutes": expected_period_minutes,
    }


def google_vendor_state(
    google_user: dict[str, Any] | None,
) -> tuple[str, dict[str, Any]]:
    """Vendor state + evidence for one leaver's Google account."""
    if google_user is None:
        return VSTATE_ABSENT, {}
    ev = {
        "last_login_at": google_user.get("lastLoginTime"),
        "org_unit": google_user.get("orgUnitPath"),
        "is_admin": bool(google_user.get("isAdmin")),
        "aliases": google_user.get("aliases") or [],
    }
    if google_user.get("suspended"):
        return VSTATE_DISABLED, ev
    return VSTATE_ACTIVE, ev


def atlassian_vendor_state(
    account: dict[str, Any] | None,
) -> tuple[str, dict[str, Any]]:
    """Vendor state + evidence for one leaver's Atlassian account.

    ``closed`` (the person closed their own Atlassian account) counts as
    disabled: nobody can sign in to it either.
    """
    if account is None:
        return VSTATE_ABSENT, {}
    status = str(account.get("account_status") or "").strip().lower()
    ev = {
        "account_status": status or None,
        "access_billable": account.get("access_billable"),
        "product_access": [
            str(p.get("key") or p.get("name"))
            for p in (account.get("product_access") or [])
            if isinstance(p, dict)
        ][:12],
    }
    if status in ("inactive", "closed", "deactivated", "suspended"):
        return VSTATE_DISABLED, ev
    if status == "active":
        return VSTATE_ACTIVE, ev
    return VSTATE_UNKNOWN, ev


def helix_action_for_google(session_enforcement: Any) -> str:
    """Condense the offboarding ledger into one word for the table.

    Deliberately reports ``no_process`` rather than blank for people who
    were offboarded before the flow existed: "we never ran anything for
    this person" is a different, actionable answer from "we ran it and it
    did nothing", and collapsing both into an empty cell is how the
    second one hides.
    """
    se = session_enforcement if isinstance(session_enforcement, dict) else None
    if not se:
        return "no_process"
    deletion = se.get("google_account_deletion")
    if isinstance(deletion, dict) and deletion.get("status"):
        return str(deletion["status"])
    acct = se.get("google_account")
    if isinstance(acct, dict) and acct.get("status"):
        return str(acct["status"])
    return "no_record"


# ─────────────────────────────────────────────────────────────────
# Read side
# ─────────────────────────────────────────────────────────────────
#
# The page groups by PERSON, not by row. "王迪 的 Google 还活着、Atlassian
# 也还活着" is one problem a person owns end-to-end; showing it as two
# unrelated table rows makes the reader do the join, and a governance
# surface that makes the reader do the join gets skimmed instead of used.

# Worst-first, because the only ordering a governance list should ever
# have is "what do I do next".
_VERDICT_RANK = {
    VERDICT_AT_RISK: 0,
    VERDICT_EXEMPT: 1,
    VERDICT_UNKNOWN: 2,
    VERDICT_OK: 3,
}


def _row_public(row: OffboardingAccountAssurance) -> dict[str, Any]:
    return {
        "system": row.system,
        "vendor_state": row.vendor_state,
        "vendor_account_id": row.vendor_account_id,
        "helix_action": row.helix_action,
        "verdict": row.verdict,
        "exempt_reason": row.exempt_reason,
        "detail": row.detail or {},
        "evidence_at": row.evidence_at.isoformat() if row.evidence_at else None,
        "at_risk_since": (
            row.at_risk_since.isoformat() if row.at_risk_since else None
        ),
    }


def _person_verdict(systems: list[dict[str, Any]]) -> str:
    """A person is as closed as their most-open system."""
    if not systems:
        return VERDICT_UNKNOWN
    worst = min(
        (str(s["verdict"]) for s in systems),
        key=lambda v: _VERDICT_RANK.get(v, 99),
    )
    return worst


async def list_people(
    db: AsyncSession,
    *,
    verdict: str | None = None,
    system: str | None = None,
    q: str | None = None,
) -> list[dict[str, Any]]:
    """Every leaver with at least one assurance row, worst first."""
    from app.identity.person_models import IdentityPerson

    stmt = select(OffboardingAccountAssurance)
    if system:
        stmt = stmt.where(OffboardingAccountAssurance.system == system)
    rows = (await db.execute(stmt)).scalars().all()
    if not rows:
        return []

    people = {
        p.id: p
        for p in (
            await db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.id.in_({r.person_id for r in rows}),
                )
            )
        ).scalars().all()
    }

    grouped: dict[Any, list[OffboardingAccountAssurance]] = {}
    for r in rows:
        grouped.setdefault(r.person_id, []).append(r)

    needle = (q or "").strip().lower()
    out: list[dict[str, Any]] = []
    for pid, prows in grouped.items():
        person = people.get(pid)
        if person is None:
            # Person hard-deleted between sweep and read — the row is a
            # fossil; skip rather than render a nameless finding.
            continue
        systems = sorted(
            (_row_public(r) for r in prows), key=lambda s: s["system"],
        )
        overall = _person_verdict(systems)
        if verdict and overall != verdict:
            continue
        if needle:
            hay = " ".join(
                str(x or "").lower() for x in (
                    person.display_name, person.work_email,
                    person.employee_no,
                )
            )
            if needle not in hay:
                continue
        at_risk_since = min(
            (s["at_risk_since"] for s in systems if s["at_risk_since"]),
            default=None,
        )
        out.append({
            "person_id": str(pid),
            "employee_no": person.employee_no,
            "display_name": person.display_name,
            "work_email": person.work_email,
            "termination_date": (
                person.termination_date.isoformat()
                if person.termination_date else None
            ),
            "verdict": overall,
            "at_risk_since": at_risk_since,
            "systems": systems,
        })
    out.sort(
        key=lambda r: (
            _VERDICT_RANK.get(r["verdict"], 99),
            r["termination_date"] or "",
        ),
    )
    return out


async def summary(db: AsyncSession) -> dict[str, Any]:
    """KPI row + chain health for the assurance page header.

    ``sweep_stale_hours`` is deliberately part of the summary: a page
    whose numbers are three days old is more dangerous than a page with
    no numbers, because it looks like an answer. The header says how old
    the evidence is, always.
    """
    rows = (
        await db.execute(select(OffboardingAccountAssurance))
    ).scalars().all()
    per_system: dict[str, dict[str, int]] = {}
    people_verdicts: dict[Any, list[str]] = {}
    newest_evidence: datetime | None = None
    for r in rows:
        bucket = per_system.setdefault(
            r.system, {v: 0 for v in _VERDICT_RANK},
        )
        bucket[r.verdict] = bucket.get(r.verdict, 0) + 1
        people_verdicts.setdefault(r.person_id, []).append(r.verdict)
        if r.evidence_at and (
            newest_evidence is None or r.evidence_at > newest_evidence
        ):
            newest_evidence = r.evidence_at

    totals = {v: 0 for v in _VERDICT_RANK}
    for verdicts in people_verdicts.values():
        worst = min(verdicts, key=lambda v: _VERDICT_RANK.get(v, 99))
        totals[worst] = totals.get(worst, 0) + 1

    beats = (
        await db.execute(select(IdentityChainHeartbeat))
    ).scalars().all()
    chains = [
        {
            "chain_key": b.chain_key,
            "status": b.status,
            "last_beat_at": (
                b.last_beat_at.isoformat() if b.last_beat_at else None
            ),
            "checked_at": b.checked_at.isoformat() if b.checked_at else None,
            "expected_period_minutes": b.expected_period_minutes,
            "detail": b.detail or {},
        }
        for b in sorted(beats, key=lambda x: x.chain_key)
    ]

    evidence_age_hours = None
    if newest_evidence is not None:
        ref = newest_evidence
        if ref.tzinfo is None:
            ref = ref.replace(tzinfo=UTC)
        evidence_age_hours = round(
            (_now() - ref).total_seconds() / 3600.0, 1,
        )

    return {
        "people_total": len(people_verdicts),
        "at_risk": totals.get(VERDICT_AT_RISK, 0),
        "exempt": totals.get(VERDICT_EXEMPT, 0),
        "unknown": totals.get(VERDICT_UNKNOWN, 0),
        "ok": totals.get(VERDICT_OK, 0),
        "per_system": per_system,
        "chains": chains,
        "evidence_at": (
            newest_evidence.isoformat() if newest_evidence else None
        ),
        "evidence_age_hours": evidence_age_hours,
    }


# ─────────────────────────────────────────────────────────────────
# Identity-sync rosters (2026-09-05, PR3)
# ─────────────────────────────────────────────────────────────────
#
# Google and Atlassian each got an assurance sweep because each had its
# own incident. Every OTHER identity system Helix pushes offboards to —
# 汇联易, 飞书通讯录, Zoom, alicloud-sso, aws-ai, NetSuite … — had exactly
# one reader: ``identity_sync.leaver_assurance``, which RAISES on a finding
# so IT gets a failure notification and nothing appears on any page. An
# alarm with no surface is an alarm people learn to acknowledge.
#
# This reads the same vendor snapshots that job reads (refreshed nightly by
# ``identity_sync.vendor_snapshot_refresh``) and lands them in the same
# per-person table the 账号终态 page already renders, so "who can still sign
# in where" is one question with one answer across every system.

SNAPSHOT_CHAIN_PREFIX = "identity_sync.snapshot:"


def identity_sync_vendor_state(
    record: Any | None, *, termination_date: Any | None,
) -> tuple[str, dict[str, Any]]:
    """Vendor state for one leaver on one identity-sync target.

    Four answers, and the two that look alike are kept apart on purpose:

    * no record in the snapshot → ``absent`` (the account is not there).
    * record with ``is_active`` false, or a vendor status that reads
      disabled → ``disabled``.
    * record ACTIVE and the snapshot was taken strictly after the
      departure → ``active``. This is the only shape that can testify to
      a missed revocation.
    * record ACTIVE but the snapshot predates the departure → ``unknown``.
      They genuinely WERE active when it was taken; reading that as a gap
      is how ``identity_sync.leaver_assurance``'s first run produced 36
      false findings out of 48 (2026-07-17).
    """
    if record is None:
        return VSTATE_ABSENT, {}
    status = str(getattr(record, "vendor_status", None) or "").strip().lower()
    fetched = getattr(record, "fetched_at", None)
    ev: dict[str, Any] = {
        "external_key": getattr(record, "external_key", None),
        "vendor_status": status or None,
        "snapshot_at": fetched.isoformat() if fetched else None,
    }
    if getattr(record, "is_active", None) is False:
        return VSTATE_DISABLED, ev
    if status in ("inactive", "disabled", "deactivated", "suspended", "closed", "terminated"):
        return VSTATE_DISABLED, ev
    if getattr(record, "is_active", None) is None and not status:
        return VSTATE_UNKNOWN, ev
    if termination_date is not None and fetched is not None:
        snap_day = fetched.date() if hasattr(fetched, "date") else None
        if snap_day is not None and snap_day <= termination_date:
            ev["note"] = (
                "snapshot predates the departure — it cannot testify about "
                "post-departure revocation"
            )
            return VSTATE_UNKNOWN, ev
    return VSTATE_ACTIVE, ev


async def upsert_from_identity_sync_targets(
    db: AsyncSession, *, max_leavers: int = 5000,
) -> dict[str, Any]:
    """One assurance row per (leaver × revocation-enabled sync target).

    DB-only: the evidence is the stored roster snapshot, and its age is
    written as a per-target heartbeat so the page can say how old the
    answer is instead of implying it is live.
    """
    from app.domains.identity_sync.models import (  # noqa: PLC0415
        IdentitySyncTarget,
        IdentitySyncVendorRecord,
    )
    from app.identity.person_models import IdentityPerson  # noqa: PLC0415

    targets = list((await db.execute(
        select(IdentitySyncTarget).where(
            IdentitySyncTarget.deleted_at.is_(None),
            IdentitySyncTarget.enabled.is_(True),
            IdentitySyncTarget.on_offboard.in_(("deactivate", "delete")),
        ).order_by(IdentitySyncTarget.slug)
    )).scalars().all())
    leavers = list((await db.execute(
        select(IdentityPerson)
        .where(
            IdentityPerson.employment_status == "left",
            IdentityPerson.deleted_at.is_(None),
        )
        .order_by(IdentityPerson.termination_date.desc())
        .limit(max_leavers)
    )).scalars().all())
    exemptions = await registry_exemptions(
        db, [(p.work_email or "").strip().lower() for p in leavers],
    )
    out: dict[str, Any] = {
        "targets": 0, "leavers": len(leavers), "rows": 0,
        "at_risk": 0, "exempt": 0, "ok": 0, "unknown": 0,
        "per_target": [], "errors": [],
    }
    if not targets or not leavers:
        return out

    for target in targets:
        system = identity_sync_assurance_system(target.slug)
        if system is None:
            out["errors"].append(
                f"{target.slug}: slug too long for the assurance system column "
                "— rename the target or widen identity_system"
            )
            continue
        records = list((await db.scalars(
            select(IdentitySyncVendorRecord).where(
                IdentitySyncVendorRecord.target_id == target.id,
                IdentitySyncVendorRecord.record_type == "user",
            ),
        )).all())
        newest = max(
            (r.fetched_at for r in records if r.fetched_at), default=None,
        )
        by_person = {
            r.matched_person_id: r for r in records if r.matched_person_id
        }
        rows: list[dict[str, Any]] = []
        for person in leavers:
            rec = by_person.get(person.id)
            # No snapshot at all for this target is NOT "the account is
            # gone" — it is "we have not looked". Absent is only earned
            # once a roster was actually read.
            if rec is None and newest is None:
                state, evidence = VSTATE_UNKNOWN, {"note": "no roster snapshot for this target"}
            else:
                state, evidence = identity_sync_vendor_state(
                    rec, termination_date=person.termination_date,
                )
            email = (person.work_email or "").strip().lower()
            exemption = exemptions.get(email)
            verdict, exempt_reason = verdict_for(state, exemption=exemption)
            detail: dict[str, Any] = {
                **evidence,
                "target_slug": target.slug,
                "target_kind": target.kind,
                "on_offboard": target.on_offboard,
                "work_email": email,
                "employee_no": person.employee_no,
            }
            if exemption:
                detail["exemption"] = exemption
            rows.append({
                "person_id": person.id,
                "vendor_state": state,
                "vendor_account_id": getattr(rec, "external_key", None),
                "helix_action": (target.on_offboard or None),
                "verdict": verdict,
                "exempt_reason": exempt_reason,
                "detail": detail,
            })
        counts = await upsert_rows(
            db, system=system, rows=rows, evidence_at=newest,
        )
        await record_heartbeat(
            db,
            chain_key=f"{SNAPSHOT_CHAIN_PREFIX}{target.slug}",
            last_beat_at=newest,
            expected_period_minutes=24 * 60,
            detail={"target_slug": target.slug, "user_records": len(records)},
            read_ok=True,
        )
        out["targets"] += 1
        out["rows"] += len(rows)
        for key in ("at_risk", "exempt", "ok", "unknown"):
            out[key] += counts.get(key, 0)
        out["per_target"].append({
            "target": target.slug,
            "system": system,
            "at_risk": counts.get(VERDICT_AT_RISK, 0),
            "snapshot_at": newest.isoformat() if newest else None,
        })
    return out


# ─────────────────────────────────────────────────────────────────
# Enforcement (the button on the page)
# ─────────────────────────────────────────────────────────────────


class AssuranceEnforceError(Exception):
    """Refusal the caller should surface verbatim to the operator."""


class AssuranceTargetNotFound(AssuranceEnforceError):
    """The person does not exist — a 404, not a policy conflict.

    Split out because the caller maps the two to different HTTP codes,
    and an operator reading "409 conflict" for a mistyped id is being
    told the wrong thing about what went wrong.
    """


async def enforce_person(
    db: AsyncSession,
    *,
    person_id: UUID,
    systems: list[str],
    actor_user_id: UUID | None,
    # Empty string rather than ``None``: ``AuditService._coerce_trace_id``
    # accepts ``str | UUID`` and turns a blank into a fresh UUID, so the
    # runtime was always fine — the ``| None`` was an annotation the
    # callee never promised to take.
    trace_id: str = "",
    override_exemption: bool = False,
) -> dict[str, Any]:
    """Close a specific leaver's accounts now, and read the result back.

    Exists because a governance page whose only verb is "look" trains
    people to stop looking. The row that says "this person can still sign
    in" should be actionable from where it is read.

    Two rules that are not negotiable here:

    * **Refuses on a person who is not a leaver.** The one truly
      unrecoverable outcome of this button is locking out somebody who
      still works here — a rehire whose HRIS status flipped back, or a
      mis-typed id. The check is against live person state, not against
      the (possibly stale) assurance row.
    * **Refuses on an account with a retention decision unless the caller
      explicitly overrides.** Same reasoning as the sweep's auto-suspend
      split: an exemption is somebody's decision, and the platform does
      not get to quietly reverse it. The override is a separate,
      deliberate argument so it cannot happen by accident.

    Idempotent: suspending an already-suspended account is a no-op at
    both vendors, so a double click costs nothing.
    """
    from app.identity.person_models import IdentityPerson

    person = await db.get(IdentityPerson, person_id)
    if person is None or person.deleted_at is not None:
        raise AssuranceTargetNotFound("person not found")
    if person.employment_status != "left":
        raise AssuranceEnforceError(
            "refusing to close accounts for someone whose employment status "
            f"is {person.employment_status!r} — this action is for leavers "
            "only. If this person really has left, fix their HRIS status "
            "first so every other control agrees."
        )

    email = (person.work_email or "").strip().lower()
    exemption = (await registry_exemptions(db, [email])).get(email)
    if exemption and not override_exemption:
        raise AssuranceEnforceError(
            "a retention decision covers this account "
            f"(external_account_registry: {exemption.get('classification')})"
            " — Helix will not silently reverse it. Remove or re-classify "
            "the registry row if the decision no longer holds, then retry."
        )
    results: dict[str, Any] = {}
    for system in systems:
        if system == ASSURANCE_SYSTEM_ATLASSIAN:
            results[system] = await _enforce_atlassian(
                db, person_id=person_id, email=email,
            )
        elif system == ASSURANCE_SYSTEM_GOOGLE:
            results[system] = await _enforce_google(db, person=person)
        elif system.startswith(ASSURANCE_SYSTEM_IDENTITY_SYNC_PREFIX):
            results[system] = await _enforce_identity_sync(
                db, person=person,
                slug=system[len(ASSURANCE_SYSTEM_IDENTITY_SYNC_PREFIX):],
            )
        else:
            results[system] = {"status": "unsupported_system"}

    from app.audit.service import AuditService

    await AuditService(db).log(
        trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id,
        action="offboarding.account_assurance_enforced",
        status="success", resource_type="identity_person",
        resource_id=str(person_id),
        detail={
            "work_email": email,
            "systems": systems,
            "results": results,
            "override_exemption": override_exemption,
            "exemption": exemption,
        },
    )
    return {"person_id": str(person_id), "results": results}


async def _resolve_atlassian_account_id(
    db: AsyncSession, driver: Any, *, person_id: UUID, email: str,
) -> tuple[str | None, str]:
    """``(account_id, how_we_found_it)`` — cheapest source first.

    Cost matters here in a way it does not in a sweep.
    ``read_org_user_state`` looks like a point read but is a **full org
    roster pull** underneath: ~1900 accounts over ~39 paginated
    requests, because the Atlassian Admin API has no by-email lookup.
    One button click was issuing that twice — around 78 HTTP calls, on a
    shared org-wide credential, to suspend one account.

    So:
      1. the assurance row the daily sweep already wrote (0 calls) —
         and the id is stable, so staleness cannot mis-target;
      2. the SITE user search (1 call);
      3. the org roster (last resort). Still needed: Atlassian's GDPR
         behaviour omits ``emailAddress`` for managed and deactivated
         accounts on the site API, which is exactly the population this
         function is asked about, so step 2 legitimately misses.
    """
    row = (
        await db.execute(
            select(OffboardingAccountAssurance).where(
                OffboardingAccountAssurance.person_id == person_id,
                OffboardingAccountAssurance.system
                == ASSURANCE_SYSTEM_ATLASSIAN,
            )
        )
    ).scalars().first()
    if row is not None and row.vendor_account_id:
        return row.vendor_account_id, "assurance_row"
    try:
        found = await driver.find_user_by_email(email)
    except Exception as exc:  # noqa: BLE001 — fall through to the roster
        logger.info(
            "assurance: site user search failed for %s (%s) — falling back "
            "to the org roster", email, exc,
        )
        found = None
    if isinstance(found, dict) and found.get("accountId"):
        return str(found["accountId"]), "site_search"
    state = await driver.read_org_user_state(email)
    if state and state.get("account_id"):
        return str(state["account_id"]), "org_roster"
    return None, "not_found"


async def _enforce_atlassian(
    db: AsyncSession, *, person_id: UUID, email: str,
) -> dict[str, Any]:
    from app.core.coordinators.jira_seat_coordinator import load_jira_driver

    driver = await load_jira_driver(db)
    if driver is None:
        return {"status": "skipped", "reason": "jira integration not configured"}
    if not driver._org_id() or not driver._org_unscoped_api_key():
        return {
            "status": "skipped",
            "reason": (
                "org_id / org_unscoped_api_key not configured — the "
                "directory suspend-access endpoint needs an UNSCOPED org "
                "API key (create one at admin.atlassian.com with no scopes)"
            ),
        }
    account_id, via = await _resolve_atlassian_account_id(
        db, driver, person_id=person_id, email=email,
    )
    if not account_id:
        return {"status": "already_absent", "reason": "no managed account"}
    res = await driver.set_org_user_access_suspended(
        str(account_id), suspended=True,
    )
    if not res.get("ok"):
        return {
            "status": "failed",
            "resolved_via": via,
            "http_status": res.get("status"),
            "reason": res.get("error"),
        }
    # **No read-back here, deliberately.** The only org-level read
    # available is the aggregated roster, which Atlassian refreshes with
    # a lag of up to ~24h — so a read-back immediately after a successful
    # suspend cannot distinguish "it did not work" from "not visible
    # yet", and would spend another ~39 requests to say so. Claiming
    # ``verified: false`` off that would be worse than saying nothing:
    # an operator would go re-suspend an account that is already closed.
    # The daily sweep re-reads vendor truth and settles it, which is the
    # honest place for verification to live.
    # A 404 in the suspend direction is idempotent success at the driver
    # (no managed account = cannot sign in), but it is NOT the same story
    # for an operator: it means the id we aimed at is not there. That
    # happens when the cached id is stale — the account was deleted and
    # recreated at the vendor — and reporting it as a plain "suspended"
    # would let somebody close a ticket on an account nobody touched.
    if res.get("already_absent"):
        return {
            "status": "already_absent",
            "resolved_via": via,
            "http_status": res.get("status"),
            "verified": None,
            "note": (
                "no managed account under this org for that id — nothing "
                "to suspend (the id may be stale; the daily sweep re-reads "
                "it from the vendor)"
            ),
        }
    return {
        "status": "suspended",
        "resolved_via": via,
        "http_status": res.get("status"),
        "verified": None,
        "note": (
            "vendor listing lags up to ~24h — the daily assurance sweep "
            "confirms and the row turns green then"
        ),
    }


async def _enforce_google(db: AsyncSession, *, person: Any) -> dict[str, Any]:
    from app.core.coordinators import account_lifecycle_coordinator

    res = await account_lifecycle_coordinator.suspend_google_account(db, person)
    return {
        "status": res.get("status"),
        "already_absent": res.get("already_absent"),
        "reason": res.get("error"),
    }


async def _enforce_identity_sync(
    db: AsyncSession, *, person: Any, slug: str,
) -> dict[str, Any]:
    """Close this leaver's account on ONE identity-sync target, now.

    Reuses ``IdentitySyncService.revoke_leavers_on_target`` — the write
    sibling the nightly reconcile already uses — rather than a second
    revoke path: two ways to revoke would eventually disagree about which
    populations are in scope, and this one already handles the 2061 and
    orphan cases plus the runner capability gate. Scoped to this person by
    filtering the target's candidate set afterwards, so the button never
    silently revokes a batch the operator did not ask for.
    """
    from app.domains.identity_sync.models import (  # noqa: PLC0415
        IdentitySyncTarget,
    )
    from app.domains.identity_sync.service import (  # noqa: PLC0415
        IdentitySyncService,
    )

    target = (await db.execute(
        select(IdentitySyncTarget).where(
            IdentitySyncTarget.slug == slug,
            IdentitySyncTarget.deleted_at.is_(None),
        )
    )).scalars().first()
    if target is None:
        return {"status": "target_not_found", "reason": slug}
    svc = IdentitySyncService(db)
    pairs = await svc.active_leaver_vendor_records(
        target_id=target.id, limit=1000,
    )
    mine = [(p, v) for p, v in pairs if p.id == person.id]
    if not mine:
        # Nothing the shared Signal-B population says is open for this
        # person on this target. Reported as a no-op rather than a
        # success: "there was nothing to do" and "I closed it" are
        # different answers and the operator is entitled to the true one.
        return {"status": "nothing_to_revoke", "target": slug}
    result = await svc.revoke_leavers_on_target(target, person_id=person.id)
    status = (
        "revoked" if result.get("revoked")
        else "failed" if result.get("failed")
        else "skipped"
    )
    return {"status": status, "target": slug, **result}