"""Offboarding + asset transfer domain models.

OffboardingProcess — one row per person who is or was leaving. Created
when HRIS sync confirms an offboard event. Transitions:

    scanning   — just created; service is enumerating blockers
    blocked    — ≥1 blocker with resolution='pending'
    ready      — all blockers resolved; SSO can be suspended
    completed  — SSO suspended + all connectors executed
    cancelled  — admin aborted (e.g., HRIS flipped back to active)

OffboardingBlocker — a single active resource (hardware, seat, permission,
api_key, cloud_account) that must be handled before the process is "ready".
Links to the provisioning_task or transfer_request created to resolve it.

AssetTransferRequest — generic peer-to-peer handoff. Used both during
offboarding (leaver transfers to colleague) and regular operations.
"""
from __future__ import annotations

import uuid
from datetime import date as date_type
from datetime import datetime
from typing import Any

from sqlalchemy import (
    JSON,
    Date,
    DateTime,
    ForeignKey,
    Integer,
    String,
    Text,
    UniqueConstraint,
    func,
)
from sqlalchemy.orm import Mapped, mapped_column

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

# Process statuses
PROC_SCANNING = "scanning"
PROC_BLOCKED = "blocked"
PROC_READY = "ready"
PROC_COMPLETED = "completed"
PROC_CANCELLED = "cancelled"

# Blocker resource types
RES_HARDWARE = "hardware"
RES_SOFTWARE_SEAT = "software_seat"
RES_SOFTWARE_PERMISSION = "software_permission"
RES_API_KEY = "api_key"
# ``RES_CLOUD_ACCOUNT`` (legacy cmdb_cloud_accounts row pointer) was
# removed in migration 053 — use RES_CLOUD_SCOPE_OWNER and
# RES_CLOUD_ACCESS_GRANT instead.
# Network-asset ownership (PR unified-connectors). On leaver these force-
# transfer to the manager cascade (not revoked) — these are registered
# domain / DNS records / cloud servers whose "owner" is the registrant
# or responsible admin. Keeping the resource alive is important;
# only the owner FK needs to move.
RES_DOMAIN = "domain"
RES_DNS_RECORD = "dns_record"
RES_CLOUD_SERVER = "cloud_server"
# LB + CDN owner force-transfer (PR #346, 2026-05-18). Migration 173 +
# 160 gave both tables ``owner_person_id`` FKs but the offboarding
# blocker scanner only walked domain/DNS/server. Without these, a
# leaver kept perpetual ownership of every LB and CDN domain they
# touched — silently orphaned every time the offboarding force-
# transfer "completed" without addressing them.
RES_LOAD_BALANCER = "load_balancer"
RES_CDN_DOMAIN = "cdn_domain"
# Software catalog entries where the leaver is listed as the
# business ``owner_person_id`` or ``it_contact_person_id``. Always
# force-transfer up the manager cascade — removing the FK would leave
# orphan software entries that no-one is responsible for, and SSO
# cascade has no bearing on these (they describe the software, not
# the leaver's access to it).
RES_SOFTWARE_OWNER = "software_owner"
RES_SOFTWARE_IT_CONTACT = "software_it_contact"
# Google Workspace mail groups where the leaver is listed as an owner.
# These MUST be transferred to someone else before the leaver's
# account is suspended — Google requires at least one owner per
# group, and SSO cascade does NOT auto-reassign ownership.
RES_EMAIL_GROUP = "email_group"
# Multi-cloud scope ownership (PR 2026-04-23). If the leaver owns a
# cloud scope (AWS account / Azure sub / GCP project / Alicloud RAM
# member), the scope MUST be transferred to someone else before the
# leaver can be marked ``left`` — otherwise the scope enters the
# unowned-compliance-incident state and blocks new grants.
RES_CLOUD_SCOPE_OWNER = "cloud_scope_owner"
# Generic-flow resources (added 2026-05-02): the leaver flow demands a
# named successor before the row clears, mirroring the pattern used by
# RES_CLOUD_SCOPE_OWNER. Both are MUST_TRANSFER — they aren't
# auto-revokable because the SSO seat / registered domain remains a
# real org resource that needs an attached human.
RES_SSO_GRANT = "sso_grant"
RES_DOMAIN_REGISTRATION = "domain_registration"
# Internal system registry ownership (``registry_systems.owner_person_id``,
# added 2026-08-25). If the leaver is the named owner of a live
# first-party system, that ownership MUST move before the departure
# completes: the system keeps running and keeps receiving inbound
# vulnerability reports, and a report routed to a departed owner produces
# silence rather than a visible unassigned queue item. MUST_TRANSFER,
# never revoke — there is nothing to revoke, only someone to name.
#
# The registry records a backup owner precisely so this handover has an
# obvious successor; the T+0+24h force transfer prefers them over the
# generic manager fallback.
RES_SYSTEM_OWNER = "system_owner"
# Approval-flow registry ownership (``registry_approval_flows.owner_person_id``,
# added 2026-08-26). The flow keeps running in Feishu whatever we do here;
# what stops is anyone answering for it. The all-staff page publishes that
# name, so a departed owner turns 「这个流程找谁」 into a dead end for the
# whole company. MUST_TRANSFER — there is nothing to revoke, only somebody
# to name. Unlike the system registry there is no backup-owner slot to
# prefer, so the T+0 force transfer falls through to the manager.
RES_APPROVAL_FLOW_OWNER = "approval_flow_owner"
# Multi-cloud access grants the leaver holds. These are always revoked
# (not transferred) — when a person leaves they should lose cloud
# access; the next person joining the role can re-request.
RES_CLOUD_ACCESS_GRANT = "cloud_access_grant"
# Google Workspace orphan accounts (shared mailboxes / service accounts)
# the leaver is the responsible "owner" for. MUST_TRANSFER — the
# account itself stays alive (it's a real, billed mailbox doing real
# work), only the owner FK moves. Without a re-assignment the leaver
# flow stays blocked, which is the whole point: every orphan needs a
# named human at all times.
RES_GOOGLE_ORPHAN_OWNER = "google_orphan_owner"
# Cloud IAM user ownership (cmdb_iam_users.owner_person_id). When a Helix
# admin maps an IAM user (AWS IAM / Aliyun RAM / Tencent CAM) to a Helix
# person, that person is on the hook for the IAM credential — rotations,
# revocations, audit responses. On offboarding the FK MUST move to a
# replacement human (manager fallback). The actual IAM credential stays
# alive on the cloud side; only the Helix-side ownership pointer moves.
# Before round-20 (2026-05-12) this gap was only caught by the T+1
# post-leaver verifier, which raised a compliance_incident rather than
# resolving it — leaving leavers' IAM users permanently mapped until an
# admin acted manually.
RES_CLOUD_IAM_USER = "cloud_iam_user"
# VPN application access (Feilian / corporate VPN). Approved entitlements
# MUST be revoked on offboarding — without this scan a leaver can
# continue to use overseas-resource access from corporate VPN past
# their leave date, until an admin manually clicks revoke. Auto-revoke
# is the right default; the admin path already exists at
# ``/vpn-applications/{id}/revoke``.
RES_VPN_APPLICATION = "vpn_application"
# AI tool applications (ChatGPT / Bailian seats granted via Helix).
# Same shape as VPN: approved seats stay "approved" for a leaver
# indefinitely without an offboarding sweep, leaking budget and seat
# allocations.
RES_AI_TOOL_APPLICATION = "ai_tool_application"
# Provider account-wide access revoke (currently Jira/Atlassian). A
# leaver's product access can come from arbitrarily many groups
# (jira-administrators, SOC2, per-site product groups, …), so removing
# them from one bound group never covers everything — the reliable
# action is removing them from EVERY group at once (the account itself
# is preserved; Helix policy never deactivates the managed account so
# issue history stays attributed). This blocker is created off the
# provider integration's existence (not off tracked seats), so a leaver
# whose access lived only in untracked groups is still caught. Auto-
# revoke handler = driver.revoke_all_access (idempotent), executed via a
# target_type='account' provisioning task.
RES_SOFTWARE_ACCOUNT = "software_account"
# Identity-consumer ownership (identity_consumers.owner_person_id). The
# leaver is the accountable human for an external system that pulls
# directory data (PII egress, PLATFORM §10.6). On offboarding the owner
# can be handed to a named successor (the consumer stays live under the
# new owner); if NOBODY claims it by T+0 the auto-handler DISABLES the
# consumer (kills the egress) rather than silently reassigning it to a
# manager who never asked for it — a PII data tap must not run under a
# departed (or unwitting) owner. Distinct from the cloud-scope-owner
# MUST_TRANSFER pattern precisely because the safe default here is
# "stop", not "reassign".
RES_IDENTITY_CONSUMER_OWNER = "identity_consumer_owner"
# GitHub 组织成员资格。**人绑权限**,离职时在源头撤销,不需要合成一条
# transfer 记录 —— 与 sso_grant / vpn_application / cloud_access_grant 同族,
# 所以它进 ``IMMEDIATE_REVOKE_RESOURCE_TYPES``(T+0 第一个 tick 就撤,不吃
# 24h grace)。
#
# 这条资源类型存在之前,离职流程对 GitHub **完全没有概念**:一个离开
# 公司的人对 5 个组织(pingcap / tikv / pingcap-inc / PingCAP-QE /
# tidbcloud)里私有仓库的访问权,靠的是有没有人记得手工去踢。
#
# ``resource_id`` 是 ``github_org_memberships.id`` —— 一条成员资格一行
# blocker,而不是「一个人一行」:每个组织的离职策略可以不同(社区组织
# 可以豁免),合并成一行就无法表达「移出了 3 个、留了 1 个」。
#
# 两类**不会**产生 blocker 的成员资格,理由各不相同:
#   * 组织的 ``offboarding_policy='exempt'`` —— 社区组织,离职的人可能
#     仍是合法的 maintainer。豁免是 org owner 明确做出的决定。
#   * 台账上 ``person_id`` 为空的记录 —— Helix 认不出它属于谁,所以它
#     根本不会出现在任何人的离职流程里。这不是遗漏,是这套东西已知的
#     边界:它由对账面的「认不出是谁」计数负责,不是由离职流程负责。
RES_GITHUB_ORG_MEMBERSHIP = "github_org_membership"
# DAX 个人数据集授权(migration 606,2026-09-05)。``data_access_bindings`` 里
# ``principal_type_snapshot='google_user'`` 且 ``principal_subject_snapshot`` 是这个人
# 的 active 行 —— 一条 BigQuery 数据集的个人读权限。人绑权限,在源头撤销:走 DAX
# 自己的 ``revoke_data_access_binding``(system_context)→ apply-absent reconciler,
# 所以 BigQuery 仍只有一个写入者。进 ``IMMEDIATE_REVOKE_RESOURCE_TYPES``(T+0 第一
# 个 tick 就撤)。此前离职流程对 DAX 完全没有概念:离职者的数据集访问只靠
# ``revoke_stale_personal_bindings`` 在下一次 apply 里按 employment_status 兜底 ——
# 那是安全网,不是流程。``resource_id`` 是 ``data_access_bindings.id``。
RES_DATA_ACCESS_GRANT = "data_access_grant"
# ── 2026-09-05(访问对象闭环 PR3,迁移 609)—— 三个此前离职流程「看不见」的持有 ──
# 对象存储桶负责人(``cmdb_object_storage.owner_person_id``)。与 LB / CDN 同族:
# Helix 侧的负责人 FK 转给交接人,桶本身不动。
RES_OBJECT_STORAGE = "object_storage"
# TLS 证书负责人(``cmdb_certificates.owner_person_id``)—— **只算 ``owner_is_manual``
# 的行**:探测 / 云同步来的证书负责人是实时从 DNS/域名链解析的,不是这个人的持有,
# 转它等于跟解析器打架。手动指定的负责人才是「这个人担着」的事。
RES_CERTIFICATE = "certificate"
# Google 邮件组的**静态成员资格**。此前离职流程只处理组的所有权(``RES_EMAIL_GROUP``);
# 一个离职者作为普通成员留在 hr-all@ / finance-team@ 里,Google 账号停用后邮件不再
# 投递,但账号被保留(CBG 保留 / 共享服务)或复活时,组里的那一行仍然给他寄内部邮件。
# 人绑权限,T+0 在源头 ``members.delete``;``resource_id`` 是本地 ``email_groups.id``
# (Google 名册里 Helix 未镜像的组只计入扫描摘要,不造 blocker)。动态组不算 —— 成员
# 关系由规则算出,人走了规则自然不再命中。
RES_EMAIL_GROUP_MEMBER = "email_group_member"

# Blocker resolutions
RESO_PENDING = "pending"
RESO_TRANSFERRED = "transferred"
RESO_REVOKED_AUTO = "revoked_auto"
RESO_REVOKED_MANUAL = "revoked_manual"
RESO_SKIPPED = "skipped"
# Resource will be taken care of by SSO cascade (Google/Feishu suspend).
# Auto-set by the scanner when the per-connector policy is ``sso_cascade``
# — distinct from ``skipped`` because it carries an explicit audit trail
# that the platform has checked the resource and chosen to not act.
RESO_SSO_CASCADE = "sso_cascade"
# Resource is the identity itself (a Feishu / Google account, not a SaaS
# seat behind it). Its lifecycle is owned by an Identity Sync Center
# target (``feishu-contact`` / future ``google-contact``) that pushes
# HRIS → vendor on hire and PATCH-freezes on offboard. The asset/seat
# layer never creates a provisioning task for these — duplicating the
# call from two paths produces 409s on the vendor side and confuses the
# audit trail. Marker carries the explicit audit row that the platform
# saw the seat and deferred to identity_sync.
RESO_IDENTITY_SYNC_MANAGED = "identity_sync_managed"

# Round-14 severity split (2026-05-08). Splits the blocker queue into
# two display buckets so IT only sees real exceptions in the red badge:
#
#   blocker  - needs a human decision (no auto-handler, or fallback
#              failed). Default for unknown / new resource types so a
#              freshly added scanner branch surfaces in red rather than
#              silently slipping past - fail-closed by design.
#   advisory - every known auto-handler covers this resource_type.
#              Shown under "system-will-auto-handle" with the T+0 grace
#              countdown; IT can intervene early but does not have to.
#
# UI hint only: LeaverBlockGuard and ProcessAdvanceService keep their
# original strict semantics (every pending row is force-resolved at
# T+0+24h regardless of severity), and _recompute_status still flips a
# process to PROC_BLOCKED on any pending row.
SEVERITY_BLOCKER = "blocker"
SEVERITY_ADVISORY = "advisory"

# Per-connector offboarding policies (values stored in
# ``software_connectors.offboarding_policy`` and
# ``cmdb_cloud_accounts.offboarding_policy``).
POLICY_AUTO_REVOKE = "auto_revoke"
POLICY_SSO_CASCADE = "sso_cascade"
POLICY_MUST_TRANSFER = "must_transfer"  # cloud accounts only
# The catalog row IS an identity (a Feishu / Google account, not a
# downstream SaaS that consumes one). Lifecycle is owned by an
# Identity Sync Center target — ``feishu-contact`` (HRIS → 飞书 push:
# create on hire, PATCH freeze on leave). Offboarding scan marks the
# resulting blocker ``identity_sync_managed`` and creates NO
# provisioning task; firing a duplicate revoke from the seat layer
# would 409 on the vendor side and split the audit trail. Distinct
# from ``sso_cascade`` (downstream SaaS that dies when the IdP closes)
# because here Helix is the proactive actor, not a passive consumer.
POLICY_IDENTITY_SYNC_MANAGED = "identity_sync_managed"
# The software is an un-integrated system (no connector, no seat sync) whose
# accounts Helix cannot deprovision programmatically — e.g. a shared org
# console (TiDB Cloud Org). Helix takes NO automated action; instead, when a
# person in one of the software's ``offboarding_config.notify_bus`` business
# units leaves, the software's ``owner_person_id`` is notified (Feishu +
# email) to manually close the departing person's account. Scope semantics:
# ``notify_bus`` empty / missing == notify on EVERY leaver; otherwise only
# leavers whose primary-membership BU (``app.identity.bu_helpers``) matches
# one of the listed names (case-insensitive). Fires once per offboarding
# process from ``dispatch_post_commit_actions`` (start / re-notify / T+0
# catch-up), NOT on the reconcile cron — deduped by an
# ``OffboardingReminderLog`` ``owner_manual_close`` row.
POLICY_OWNER_NOTIFY_MANUAL = "owner_notify_manual"

# Transfer statuses
XFER_PENDING = "pending"
XFER_ACCEPTED = "accepted"
XFER_REJECTED = "rejected"
XFER_CANCELLED = "cancelled"
XFER_COMPLETED = "completed"


# SLA breach levels (migration 090). Values track how long a process
# has been in `blocked` state past the leaver_date:
#
#   none    — not yet blocked, or blocked < 24h
#   warn    — 24h ≤ blocked < 48h: manager DM'd
#   urgent  — 48h ≤ blocked < 72h: IT admin DM'd
#   breach  — blocked ≥ 72h: critical compliance_incident raised,
#             daily digest highlights, leaver-blocked dashboard red
SLA_NONE = "none"
SLA_WARN = "warn"
SLA_URGENT = "urgent"
SLA_BREACH = "breach"


class OffboardingProcess(Base):
    __tablename__ = "offboarding_processes"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False, unique=True,
    )
    leaver_date: Mapped[date_type] = mapped_column(Date, nullable=False)
    status: Mapped[str] = mapped_column(String(32), nullable=False, default=PROC_SCANNING)

    scanned_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    ready_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    sso_suspended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # Per-channel identity-plane enforcement proof (migration 383,
    # 2026-07-04). The `_push_force_logout` path previously returned only
    # a bool and swallowed per-vendor failures, so IT had no durable answer
    # to "is this leaver's access actually cut?" — the exact question the
    # P0051 incident surfaced. This column records the outcome of each
    # session/account cutoff channel so the offboarding detail page + the
    # person-detail 离职 tab can show proof, and a failed channel can be
    # re-run and re-recorded. Shape (all keys optional):
    #   { "google_session": {"status": "ok"|"already_absent"|"failed"
    #                                   |"skipped_no_integration",
    #                         "at": <iso8601>, "error": <str|null>},
    #     "google_account": {"status": "suspended"|"already_absent"
    #                                   |"retained"|"failed"|"no_integration",
    #                         "at": <iso8601>, "error": <str|null>},
    #     "feishu_session": {"status": "no_public_api"|"ok"|"failed",
    #                         "at": <iso8601>, "error": <str|null>},
    #     "google_account_deletion": {"status": "deleted"|"already_absent"
    #                                   |"retained"|"failed"
    #                                   |"skipped_not_suspended"|"no_integration",
    #                         "at": <iso8601>, "error": <str|null>,
    #                         "already_absent": <bool>},
    #     "updated_at": <iso8601> }
    # ``google_account_deletion`` is the terminal-DELETE ledger written by the
    # ``hr.offboarding_google_account_delete`` sweep (2026-07-08 ADR); statuses
    # deleted/already_absent/retained are terminal (never retried).
    # NULL on processes that pre-date the column or never ran force-logout.
    session_enforcement: Mapped[dict[str, Any] | None] = mapped_column(JSON)

    # Live progress display fields (migration 090). Powered by the
    # WorkflowJournal at app/core/workflow_journal/. ``current_step``
    # carries the human-readable label of the most recently executed
    # phase; ``step_log`` is a compact JSON list of recent step
    # entries kept on the row itself for fast UI render without
    # joining workflow_step_events.
    current_step: Mapped[str | None] = mapped_column(String(128))
    step_log: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON)

    # SLA tracking. ``first_blocked_at`` is set when the process first
    # enters a state where leaver_date has passed but blockers remain;
    # ``sla_breach_level`` advances through warn → urgent → breach as
    # time passes. The leaver.escalation_tick handler maintains both.
    first_blocked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    sla_breach_level: Mapped[str] = mapped_column(
        String(16), nullable=False, default=SLA_NONE, server_default=SLA_NONE,
    )

    # Pointer to the WorkflowRun row that orchestrates this offboarding.
    # NULL on legacy processes that pre-date migration 090; new
    # processes always carry the FK.
    workflow_run_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    # Per-process default handover successor (migration 211, 2026-05-28).
    # Populated on ``start_process`` from the leaver's HRIS manager_id;
    # IT Admin can edit or clear from the detail page. Used to prefill
    # the per-blocker "指定接手人" modal (SSO grant / domain registration
    # / google_orphan_owner) — the actual ``successor_person_id`` is
    # still passed in the per-blocker request so audit attribution stays
    # at the blocker level. NULL means "no default — admin types every
    # time", which is fine for orgs that don't follow the "transfer to
    # direct manager" convention.
    default_handover_person_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"),
    )

    # Per-asset hardware disposition map (migration 184, 2026-05-19).
    # Populated by the offboarding-console UI before T+0 grace expires.
    # NULL ⇒ the advance handler falls back to ``recycle_to_office``
    # routing the asset to the leaver's home office. Shape:
    #   { "<asset_id>": {
    #       "action": "keep" | "recycle_to_office" | "transfer_to_person",
    #       "office_id": "<uuid|null>",   # required for recycle_to_office
    #       "person_id": "<uuid|null>",   # required for transfer_to_person
    #       "note": "<optional admin note>"
    #     }, ... }
    # Validated by HardwareDispositionRequest. See PR-#XXX (held_by_office)
    # for the why this exists.
    hardware_disposition: Mapped[dict[str, Any] | None] = mapped_column(JSON)

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


class OffboardingBlocker(Base):
    __tablename__ = "offboarding_blockers"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    process_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("offboarding_processes.id", ondelete="CASCADE"),
        nullable=False,
    )
    resource_type: Mapped[str] = mapped_column(String(32), nullable=False)
    resource_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    label: Mapped[str] = mapped_column(String(255), nullable=False)
    software_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)

    # Round-14 severity split. Default ``blocker`` so a new resource type
    # added to the scanner without an explicit map entry surfaces in the
    # red queue rather than silently falling into the auto-handled bucket.
    severity: Mapped[str] = mapped_column(
        String(16), nullable=False, default=SEVERITY_BLOCKER,
        server_default=SEVERITY_BLOCKER,
    )
    resolution: Mapped[str] = mapped_column(String(32), nullable=False, default=RESO_PENDING)
    resolution_ref: Mapped[str | None] = mapped_column(String(255))
    provisioning_task_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("provisioning_tasks.id"),
    )
    transfer_request_id: Mapped[uuid.UUID | None] = mapped_column(Uuid)
    notes: Mapped[str | None] = mapped_column(Text)

    # Migration 148 (2026-05-12): structured error storage. Pre-148 the
    # auto-handler failure path appended ``\n[T+0 自动转让失败] ...`` to
    # ``notes``, which conflated human comments with machine errors and
    # bloated the column over repeated tick retries. These three columns
    # are the source of truth for "did the auto-handler fail":
    #   * last_error_message — most recent failure summary, overwritten
    #     each retry (NOT appended).
    #   * last_error_at — timestamp of the most recent failure.
    #   * error_count — total failures since blocker creation. Used by
    #     the triage_bucket logic: >=1 means the blocker stays pending
    #     in the ``needs_it`` queue so admin sees it even though the
    #     resource severity is ``advisory``.
    last_error_message: Mapped[str | None] = mapped_column(Text)
    last_error_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
    )
    error_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default="0",
    )

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


# Triage buckets — UI-facing classification computed in
# :func:`compute_triage_bucket`. Server-side computed so frontend can
# render directly off the API response without re-deriving rules.
TRIAGE_NEEDS_IT = "needs_it"
TRIAGE_AUTO_IN_PROGRESS = "auto_in_progress"
TRIAGE_RESOLVED = "resolved"


def compute_triage_bucket(
    *,
    resolution: str,
    severity: str,
    error_count: int,
    provisioning_task_status: str | None,
    provisioning_task_execution_mode: str | None,
) -> str:
    """Decide which of the three admin-facing buckets a blocker falls in.

    Single source of truth — exported here so the offboarding service
    serializer, the leaver-blocked detail endpoint, and the admin-list
    KPI aggregator all agree.

    Rules (in priority order):

    1. ``resolution != 'pending'`` → ``resolved``. The blocker is done;
       no UI action needed.
    2. ``error_count >= 1`` → ``needs_it``. Auto-handler tried and
       failed at least once. Even though the platform will retry on
       the next tick, admin should know so they can pre-empt the
       retry storm (e.g., Slack SCIM plan-gated means every retry will
       fail until the binding is flipped to ``manual_notify``).
    3. ``provisioning_task.status == 'failed'`` → ``needs_it``. Same
       reasoning as (2) but covers the provisioning-task path
       (software_seat / software_permission) which doesn't write to
       ``error_count`` directly.
    4. ``provisioning_task.execution_mode == 'manual_notify' and
       provisioning_task.status == 'notified'`` → ``needs_it``. The
       Slack-style "we sent the IT team a message; please click in
       the vendor console" path. Until the admin marks done, the
       blocker is stuck and must surface.
    5. ``severity == 'blocker'`` → ``needs_it``. Fail-closed default
       for resource types that have no auto-handler mapped. The
       severity backfill in migration 136 stamped every known type as
       ``advisory``; anything left at the ``blocker`` default is a
       new resource type that needs explicit admin handling.
    6. Otherwise → ``auto_in_progress``. ``advisory`` blocker, no
       failures yet, no provisioning task or task is queued/in-flight.
       T+0+24h force-handler will resolve it without admin action.
    """
    if resolution != RESO_PENDING:
        return TRIAGE_RESOLVED
    if error_count >= 1:
        return TRIAGE_NEEDS_IT
    if provisioning_task_status == "failed":
        return TRIAGE_NEEDS_IT
    if (
        provisioning_task_execution_mode == "manual_notify"
        and provisioning_task_status == "notified"
    ):
        return TRIAGE_NEEDS_IT
    if severity == SEVERITY_BLOCKER:
        return TRIAGE_NEEDS_IT
    return TRIAGE_AUTO_IN_PROGRESS


class OffboardingReminderLog(Base):
    """Idempotency marker for leaver reminders and the day-0 auto-escalation.

    The manager-escalation worker runs on a schedule and evaluates
    every pending OffboardingProcess. To avoid re-sending the same
    reminder every tick, it records one row per (process_id,
    reminder_kind) here. ``reminder_kind`` values:

      * ``reminder_7d`` — 7 days before leaver_date
      * ``reminder_3d`` — 3 days before leaver_date
      * ``reminder_1d`` — 1 day before leaver_date
      * ``escalation`` — on leaver_date, pending MUST_TRANSFER blockers
        auto-raised a transfer to the leaver's manager
      * ``it_grace_window`` — round-13: the day a process is first seen
        by the worker, IT admins are pinged with the pending-blocker
        list and given 24h to act manually before the system runs
        force-transfer. Especially relevant for HRIS-late detection
        (P01121: discovered T+7 days, never received T-7/T-3/T-1) where
        jumping straight to manager fallback skipped the human window
        entirely.

    ``channels`` captures which channels were dispatched (``feishu``,
    ``email``, ``inapp``) for audit / debugging. ``detail`` holds the
    recipient list and blocker IDs affected.
    """

    __tablename__ = "offboarding_reminder_log"
    __mapper_args__ = {"eager_defaults": True}
    __table_args__ = (
        # Application path always treats (process_id, reminder_kind) as
        # unique — the cron does SELECT ... WHERE ... THEN add(...). The
        # single-flight cron lock keeps races closed today, but any
        # future code path (admin manual re-send, parallel handover
        # trigger, backfill script) writing the row directly would race
        # silently. Migration 179 enforces this at the DB level.
        UniqueConstraint(
            "process_id",
            "reminder_kind",
            name="uq_offboarding_reminder_log_process_kind",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    process_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("offboarding_processes.id", ondelete="CASCADE"),
        nullable=False,
    )
    reminder_kind: Mapped[str] = mapped_column(String(32), nullable=False)
    sent_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now(),
    )
    channels: Mapped[list[str] | None] = mapped_column(JSON)
    detail: Mapped[dict[str, Any] | None] = mapped_column(JSON)


REMINDER_7D = "reminder_7d"
REMINDER_3D = "reminder_3d"
REMINDER_1D = "reminder_1d"
REMINDER_ESCALATION = "escalation"
REMINDER_IT_GRACE = "it_grace_window"


class AssetTransferRequest(Base):
    __tablename__ = "asset_transfer_requests"
    __mapper_args__ = {"eager_defaults": True}

    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    initiator_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False,
    )
    recipient_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False,
    )
    asset_type: Mapped[str] = mapped_column(String(32), nullable=False)
    asset_ref_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False)
    asset_label: Mapped[str] = mapped_column(String(255), nullable=False)

    status: Mapped[str] = mapped_column(String(32), nullable=False, default=XFER_PENDING)
    reason: Mapped[str | None] = mapped_column(Text)
    response_note: Mapped[str | None] = mapped_column(Text)
    responded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    offboarding_process_id: Mapped[uuid.UUID | None] = mapped_column(
        Uuid, ForeignKey("offboarding_processes.id"),
    )

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


# ─────────────────────────────────────────────────────────────────
# Account end-state assurance
# ─────────────────────────────────────────────────────────────────
#
# Everything above proves "Helix revoked what it knew about". None of it
# proves the thing an auditor actually asks: **can this person still get
# in?** Those are different questions, and the 2026-08-13 outage is what
# happens when only the first one is asked — Helix's Jira revoke worked
# perfectly for 16 days while every leaver kept a live Atlassian account,
# because the org-side IdP directory sync that deactivates accounts had
# been switched off and nothing in Helix was looking at account state.
#
# ``OffboardingAccountAssurance`` is that second question, persisted: one
# row per (person, identity system) recording what the VENDOR says today,
# what Helix did, and whether those two together mean the person is
# actually locked out. Refreshed by the per-system leaver audits (which
# already pull the full vendor roster once per run), so the read surface
# is a plain DB query rather than a fan-out of live vendor calls.

# Identity systems tracked by the assurance sweep.
ASSURANCE_SYSTEM_GOOGLE = "google"
ASSURANCE_SYSTEM_ATLASSIAN = "atlassian"
ASSURANCE_SYSTEMS = (ASSURANCE_SYSTEM_GOOGLE, ASSURANCE_SYSTEM_ATLASSIAN)
# Identity-sync targets get one assurance system each (2026-09-05, PR3):
# ``isync:<target slug>``. The prefix is short on purpose — ``identity_system``
# is String(32) and a longer prefix would force truncation, which would let two
# targets whose slugs share a prefix silently merge their verdicts into one row.
# A slug that does not fit is SKIPPED with an error, never truncated.
ASSURANCE_SYSTEM_IDENTITY_SYNC_PREFIX = "isync:"
ASSURANCE_SYSTEM_MAX_LEN = 32


def identity_sync_assurance_system(slug: str) -> str | None:
    """``isync:<slug>`` — or ``None`` when the slug cannot fit the column."""
    name = f"{ASSURANCE_SYSTEM_IDENTITY_SYNC_PREFIX}{slug}"
    return name if len(name) <= ASSURANCE_SYSTEM_MAX_LEN else None

# What the vendor says about the account right now.
#   absent    — no account at the vendor (deleted, or never existed)
#   disabled  — account exists but cannot authenticate (Google suspended,
#               Atlassian inactive/closed)
#   active    — account exists AND can authenticate  ← the risk state
#   unknown   — the vendor could not be read this run
VSTATE_ABSENT = "absent"
VSTATE_DISABLED = "disabled"
VSTATE_ACTIVE = "active"
VSTATE_UNKNOWN = "unknown"

# The verdict a human acts on.
#   ok        — vendor state proves the leaver is locked out
#   at_risk   — leaver can still authenticate; nobody has decided this
#   exempt    — a deliberate retention decision covers this account
#   unknown   — vendor unreadable; treated as needing attention, never ok
VERDICT_OK = "ok"
VERDICT_AT_RISK = "at_risk"
VERDICT_EXEMPT = "exempt"
VERDICT_UNKNOWN = "unknown"
ASSURANCE_VERDICTS = (
    VERDICT_OK, VERDICT_AT_RISK, VERDICT_EXEMPT, VERDICT_UNKNOWN,
)


class OffboardingAccountAssurance(Base):
    """Per-leaver, per-system proof that the account is really closed.

    Refreshed wholesale by the system's leaver audit cron; never written
    from a request path. ``(person_id, system)`` is unique so the sweep
    can upsert idempotently.
    """

    __tablename__ = "offboarding_account_assurance"
    __table_args__ = (
        UniqueConstraint(
            "person_id", "identity_system",
            name="uq_account_assurance_person_system",
        ),
    )

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    person_id: Mapped[uuid.UUID] = mapped_column(
        Uuid, ForeignKey("identity_persons.id"), nullable=False, index=True,
    )
    # Column is ``identity_system``: ``SYSTEM`` is a RESERVED WORD in
    # MySQL 8.0.3+, so ``SELECT system FROM ...`` is a 1064 syntax
    # error whose message never mentions why. The ORM quotes
    # identifiers so application code never noticed — but this repo
    # runs hand-written read-only SQL against prod as a documented ops
    # path, and this is a governance table people will query by hand.
    # The Python attribute stays ``system`` so call sites read plainly.
    system: Mapped[str] = mapped_column(
        "identity_system", String(32), nullable=False, index=True,
    )

    # Vendor truth, read this run.
    vendor_state: Mapped[str] = mapped_column(
        String(16), nullable=False, default=VSTATE_UNKNOWN,
    )
    # Vendor-side identifier, so an operator can go straight to the
    # account in the vendor console without a second lookup.
    vendor_account_id: Mapped[str | None] = mapped_column(
        String(128), nullable=True,
    )
    # What Helix itself did (or was blocked from doing), copied off the
    # offboarding ledger — e.g. ``suspended`` / ``deleted`` / ``retained``
    # / ``skipped_not_suspended`` / ``no_process``.
    helix_action: Mapped[str | None] = mapped_column(String(48), nullable=True)

    verdict: Mapped[str] = mapped_column(
        String(16), nullable=False, default=VERDICT_UNKNOWN, index=True,
    )
    # Populated only for ``exempt``: WHICH decision is holding the account
    # open. An exemption nobody can name is not a decision, it's a gap —
    # this column is what makes the difference visible on the page.
    exempt_reason: Mapped[str | None] = mapped_column(String(64), nullable=True)

    # Free-form per-system evidence (product access held, last vendor
    # login, registry row id, …) — rendered in the row drawer.
    detail: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)

    # When the vendor was last read for this row. A stale ``evidence_at``
    # is itself a finding (the sweep stopped running), so it is surfaced
    # rather than hidden behind ``updated_at``.
    evidence_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True,
    )
    # First time this row was ever seen at_risk — drives "how long has
    # this been open", which is the number that makes people act.
    at_risk_since: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True,
    )

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


# Heartbeat statuses.
HEARTBEAT_OK = "ok"
HEARTBEAT_STALE = "stale"
HEARTBEAT_UNKNOWN = "unknown"


class IdentityChainHeartbeat(Base):
    """Liveness of an identity link Helix depends on but does not own.

    The 2026-08-13 outage was invisible for 16 days because the only
    signal that the Atlassian↔Google directory sync had stopped was the
    *absence* of something — and nothing watches for absences. This table
    turns each such dependency into a positive heartbeat with an
    expected period; a sweep that finds no beat within the period flips
    the row to ``stale`` and pages IT, whether or not anybody happens to
    be leaving that week.
    """

    __tablename__ = "identity_chain_heartbeats"

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid, primary_key=True, default=uuid.uuid4,
    )
    # Stable identifier, e.g. ``atlassian.google_directory_sync``.
    chain_key: Mapped[str] = mapped_column(
        String(64), nullable=False, unique=True,
    )
    status: Mapped[str] = mapped_column(
        String(16), nullable=False, default=HEARTBEAT_UNKNOWN,
    )
    # Newest vendor-side beat observed (NOT "when Helix last checked").
    last_beat_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True,
    )
    # When Helix last successfully read the vendor.
    checked_at: Mapped[datetime | None] = mapped_column(
        DateTime, nullable=True,
    )
    # Beats older than this are ``stale``.
    expected_period_minutes: Mapped[int] = mapped_column(
        Integer, nullable=False, default=240,
    )
    detail: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now(),
        nullable=False,
    )