"""The data-freshness tick.

The platform's hardest invariant: **when ``employment_status = 'left'``,
no row in any table points at this person.** ``ProcessAdvanceService``
is the automated enforcement mechanism.

On each 15-minute tick, for every :class:`OffboardingProcess` whose
owning person has ``termination_date <= today``:

1. **Walk all blockers.** Any blocker still ``resolution == 'pending'``
   is force-resolved without waiting for human input:
   - ``must_transfer`` / hardware / API keys / cloud accounts →
     **force_transfer_all**: rewrite the offending FK to point at
     the leaver's fallback owner (manager → manager's manager → ... →
     global IT admin → platform_settings.lifecycle.fallback_owner_email).
   - ``auto_revoke`` tasks → dispatch immediately; if the connector
     returns an error, the blocker flips to ``revoked_manual`` and we
     open a compliance incident so the admin can finish it by hand.
   - ``sso_cascade`` — already handled by suspend_sso later; no-op.

2. **Atomic transition.** In one DB transaction:
   - ``offboarding_process.status = 'ready'``, ``ready_at = now``
   - Invoke ``suspend_sso()`` — dispatches any residual revoke tasks
     and stamps ``sso_suspended_at``
   - ``identity_persons.employment_status = 'left'`` (if not already)

   Any exception rolls back the whole transaction, leaving the person
   in ``leaver_pending`` so the next tick retries. The invariant holds:
   ``status == left`` is only committed if force-transfer + SSO suspend
   both succeeded.

3. **Notify.** The displaced owner receives a Feishu + email message
   naming each resource that was force-transferred to them, so they
   can re-transfer, revoke, or archive as needed.

If cascade resolution can't find a valid fallback owner (manager chain
all ``left``, no IT admin configured, no fallback email) we do NOT
transition the person. Instead we raise a critical ``compliance_incident``
and leave the person in ``leaver_pending`` for human intervention.
"""
from __future__ import annotations

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

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.compliance.models import (
    SEVERITY_CRITICAL,
    TYPE_LEAVER_ORPHAN,
    TYPE_MANAGER_MISSING,
)
from app.compliance.service import ComplianceIncidentService
from app.core.exceptions import (
    AppException,
    format_exception_for_log,
    is_transaction_aborting_db_error,
)
from app.core.txn import best_effort
from app.core.working_hours import resolve_timezone
from app.domains.apikey.models import ApiKeyAssignment
from app.domains.asset.location_admin_sync import (
    resolve_location_primary_admins_bulk,
)
from app.domains.asset.location_resolver import resolve_location_id_by_name
from app.domains.asset.models import (
    AssetItem,
    AssetOfficeAddress,
    AssetSoftwareCatalog,
)
from app.domains.asset.office_local_admins import (
    resolve_office_location_primary_admin,
)
from app.domains.cmdb.models import (
    CmdbCdnDomain,
    CmdbCertificate,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbDnsRecord,
    CmdbDomain,
    CmdbIamUser,
    CmdbLoadBalancer,
    CmdbObjectStorage,
    CmdbServer,
)
from app.domains.leaver_blocked.guard import (
    LEAVER_VIA_PROCESS_ADVANCE,
    LeaverBlockGuard,
)
from app.domains.offboarding.models import (
    PROC_BLOCKED,
    PROC_READY,
    PROC_SCANNING,
    RES_AI_TOOL_APPLICATION,
    RES_API_KEY,
    RES_CDN_DOMAIN,
    RES_CERTIFICATE,
    RES_CLOUD_ACCESS_GRANT,
    RES_CLOUD_IAM_USER,
    RES_CLOUD_SCOPE_OWNER,
    RES_CLOUD_SERVER,
    RES_DATA_ACCESS_GRANT,
    RES_DNS_RECORD,
    RES_DOMAIN,
    RES_DOMAIN_REGISTRATION,
    RES_EMAIL_GROUP,
    RES_EMAIL_GROUP_MEMBER,
    RES_GITHUB_ORG_MEMBERSHIP,
    RES_GOOGLE_ORPHAN_OWNER,
    RES_HARDWARE,
    RES_IDENTITY_CONSUMER_OWNER,
    RES_LOAD_BALANCER,
    RES_OBJECT_STORAGE,
    RES_SOFTWARE_ACCOUNT,
    RES_SOFTWARE_IT_CONTACT,
    RES_SOFTWARE_OWNER,
    RES_SOFTWARE_PERMISSION,
    RES_SOFTWARE_SEAT,
    RES_SSO_GRANT,
    RES_SYSTEM_OWNER,
    RES_VPN_APPLICATION,
    RESO_PENDING,
    RESO_REVOKED_AUTO,
    RESO_TRANSFERRED,
    XFER_COMPLETED,
    AssetTransferRequest,
    OffboardingBlocker,
    OffboardingProcess,
)
from app.domains.offboarding.service import OffboardingService
from app.identity.person_models import IdentityPerson
from app.integrations.service import get_platform_setting
from app.notifications.service import NotificationEnvelope, NotificationService

logger = logging.getLogger(__name__)


# Round-58 (2026-05-30) — T+0 grace window scope refinement.
#
# Before this round every pending blocker — including pure permission
# grants (cloud_access_grant, sso_grant, VPN/AI app entitlements) — was
# held inside the 24h grace window. Real-world feedback: a leaver whose
# offboarding was discovered late (HRIS-late detection, common at
# PingCAP) walks out the door with live cloud roles still attached for
# a full extra day. The grace window's purpose is to give IT a window
# to *physically* hand over hardware / decide who inherits network
# ownership — NOT to leave RBAC alive past T+0.
#
# Two carve-outs:
#
# ``IMMEDIATE_REVOKE_RESOURCE_TYPES`` — people-bound permission grants.
# When a leaver is detected we run the direct-revoke branch of
# ``_handle_one_blocker`` for these on the FIRST tick, regardless of
# how long the process has been known. The grace-window IT push is
# still issued for any remaining grace-eligible blockers, but these
# resource types are already resolved by the time the push lands so
# they do NOT appear in the "still pending" list.
#
# ``SCAN_SCHEDULED_RESOURCE_TYPES`` — software seats / permissions /
# Jira full-access. ``OffboardingService._apply_software_policy`` /
# ``_apply_account_policy`` create a ``ProvisioningTask`` with
# ``not_before = leaver_date + 1 (in the leaver's local TZ)`` at scan
# time; the task pipeline owns their resolution. They MUST NOT
# participate in the advance-tick grace decision: a still-pending
# blocker here means the provisioning task is in-flight (or failed and
# being retried by ``revoke_retry_cron``), not that IT should hold
# their hand. Including them in the grace push would mis-route IT
# attention to a resource the system is already revoking on its own
# schedule.
IMMEDIATE_REVOKE_RESOURCE_TYPES: frozenset[str] = frozenset({
    RES_SSO_GRANT,
    RES_VPN_APPLICATION,
    RES_AI_TOOL_APPLICATION,
    RES_CLOUD_ACCESS_GRANT,
    # GitHub 组织成员资格(2026-08-29)。同族理由:人绑权限,在源头
    # 撤销。让它在 24h grace 窗口里活着,等于一个已经离开的人对私有
    # 仓库多留一天访问权 —— 而 grace 窗口的用途是给 IT 时间去**实物**
    # 交接硬件,不是把 RBAC 留到 T+0 之后。
    RES_GITHUB_ORG_MEMBERSHIP,
    # DAX 个人数据集授权(2026-09-05,migration 606)。同族:人绑权限,T+0 在源头
    # 撤销(DAX 自己的 revoke → apply-absent reconciler)。
    RES_DATA_ACCESS_GRANT,
    # 邮件组静态成员资格(2026-09-05,PR3)。人绑权限:留着它,一个被保留的账号
    # 仍然收内部邮件。所有权(RES_EMAIL_GROUP)是转让、成员资格是移除,两回事。
    RES_EMAIL_GROUP_MEMBER,
})

SCAN_SCHEDULED_RESOURCE_TYPES: frozenset[str] = frozenset({
    RES_SOFTWARE_SEAT,
    RES_SOFTWARE_PERMISSION,
    RES_SOFTWARE_ACCOUNT,
})


class ProcessAdvanceService:
    """Orchestrates the T+0 data-freshness tick."""

    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.offboarding = OffboardingService(db)
        self.incidents = ComplianceIncidentService(db)
        self.notify = NotificationService(db)
        self.leaver_guard = LeaverBlockGuard(db)

    async def run_tick(
        self,
        *,
        fallback_email: str | None = None,
        today: datetime | None = None,
    ) -> dict[str, Any]:
        """Advance every leaver whose ``termination_date`` has passed.

        Returns a summary dict that gets stored as
        ``scheduled_jobs.last_result_summary`` so the admin UI shows
        "processed 3 leavers, 2 transitioned, 1 blocked".
        """
        today_dt = today or datetime.now(UTC)
        today_date = today_dt.date()

        summary: dict[str, Any] = {
            "processed": 0,
            "force_transferred_blockers": 0,
            "transitioned_to_left": 0,
            # Round-13: ``grace_held`` is a leaver discovered in the
            # last 24h whose force-transfer is intentionally deferred
            # one day so IT admins have a window to act manually. NOT
            # an error — separated from ``held_missing_fallback`` so
            # the operator dashboard can distinguish "waiting on IT"
            # vs "stuck, no fallback".
            "grace_held": 0,
            # Round-22 (2026-05-13) follow-up to Round-21: the router
            # already splits "no fallback configured" from "force-
            # transfer rejected every blocker" via the ``hold_reason``
            # discriminator on ``_advance_single`` return. ``run_tick``
            # was conflating them under ``held_missing_fallback`` so
            # the operator cron summary couldn't distinguish the two
            # — sending admins to configure fallbacks that were
            # already set. Mirror the router's shape so the cron + the
            # admin-button report on the same conceptual axes.
            "held_missing_fallback": 0,
            "held_pending_blockers": 0,
            # Round-58 (2026-05-30): leavers whose only pending rows are
            # ``SCAN_SCHEDULED_RESOURCE_TYPES`` (software_seat /
            # software_permission / software_account) waiting on a
            # not-yet-fired ProvisioningTask. Distinct from
            # ``held_missing_fallback`` (a real config error operators
            # must act on) — this bucket is "system is working on it,
            # next tick will pick up cleanly".
            "held_awaiting_provisioning": 0,
            "pending_blocker_count": 0,
            "incidents_created": 0,
            "errors": [],
        }

        # Look up the fallback: platform setting > the argument.
        cfg_fallback = await get_platform_setting(
            "lifecycle.fallback_owner_email", default=None, session=self.db,
        )
        resolved_fallback = (cfg_fallback or fallback_email or "").strip() or None

        # Every leaver_pending person whose termination_date <= today.
        # SQL pre-filter is on UTC date (cheap fan-in); per-person
        # local-TZ refinement below drops leavers whose own last working
        # day hasn't yet ended in their wall-clock — matches the
        # ``SuccessFactorsLifecycleService.transition_overdue_leavers``
        # behaviour added in PR #443. Without this refinement a Los
        # Angeles leaver with ``termination_date=2026-05-21`` would get
        # force-transferred at 2026-05-21 00:00 UTC = 2026-05-20 17:00
        # PDT — i.e. before their last day even starts in their zone.
        leavers = list((
            await self.db.execute(
                select(IdentityPerson)
                .options(selectinload(IdentityPerson.manager))
                .where(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.termination_date.isnot(None),
                    IdentityPerson.termination_date <= today_date,
                    IdentityPerson.employment_status.in_(("leaver_pending",)),
                )
            )
        ).scalars().all())

        deferred_local_tz_ids: list[str] = []
        eligible_leavers: list[IdentityPerson] = []
        for _p in leavers:
            try:
                tz = resolve_timezone(_p)
            except Exception:  # noqa: BLE001 — never let TZ resolution kill the tick
                tz = None
            if tz is None:
                eligible_leavers.append(_p)
                continue
            local_today = today_dt.astimezone(tz).date()
            if _p.termination_date is not None and local_today <= _p.termination_date:
                deferred_local_tz_ids.append(str(_p.id))
                continue
            eligible_leavers.append(_p)
        if deferred_local_tz_ids:
            summary["deferred_local_tz_count"] = len(deferred_local_tz_ids)
            summary["deferred_local_tz_ids"] = deferred_local_tz_ids
        leavers = eligible_leavers

        # Persons whose employment_status flipped to 'left' in this
        # tick — we fire Identity Sync Center event hooks for them
        # AFTER the outer session commits. The hook opens a fresh
        # session, so firing inside the loop would let it read the
        # pre-flip employment_status and dispatch an upsert with
        # leaver_pending data instead of a deactivate.
        transitioned_person_ids: list = []

        for person in leavers:
            summary["processed"] += 1
            # Per-person savepoint so A's partial failure doesn't
            # leave A half-transitioned AND doesn't abort B+C+...
            # Everything inside ``_advance_single`` is atomic: either
            # force-transfer + suspend_sso + status=left all commit,
            # or all roll back.
            try:
                async with self.db.begin_nested():
                    result = await self._advance_single(
                        person, fallback_email=resolved_fallback,
                    )
                summary["force_transferred_blockers"] += result["force_transferred"]
                if result["transitioned"]:
                    summary["transitioned_to_left"] += 1
                    transitioned_person_ids.append(person.id)
                elif result.get("grace_held"):
                    summary["grace_held"] += 1
                else:
                    # Round-22 follow-up: use the ``hold_reason``
                    # discriminator from ``_advance_single`` so the
                    # cron summary distinguishes "no fallback configured"
                    # from "force-transfer rejected every blocker"
                    # (the prod RBAC FORBIDDEN case from Round-21).
                    hold_reason = result.get("hold_reason")
                    if hold_reason == "pending_blockers":
                        summary["held_pending_blockers"] += 1
                        summary["pending_blocker_count"] += result.get(
                            "pending_blocker_count", 0,
                        )
                    elif hold_reason == "awaiting_provisioning":
                        # Round-58 (2026-05-30): ProvisioningTask
                        # in-flight is a temporary hold, NOT an operator-
                        # actionable failure. Without this explicit
                        # branch the row fell into ``missing_fallback``
                        # and the admin dashboard told operators "no
                        # fallback configured" — sending them to fix a
                        # config that was never broken (Round-21's
                        # mistake all over again).
                        summary["held_awaiting_provisioning"] += 1
                        summary["pending_blocker_count"] += result.get(
                            "pending_blocker_count", 0,
                        )
                    else:
                        # ``missing_fallback`` or unknown — preserve
                        # the legacy counter so existing log-scrapers
                        # continue to find the field.
                        summary["held_missing_fallback"] += 1
                summary["incidents_created"] += result["incidents_created"]
            except Exception as exc:  # noqa: BLE001
                # Transaction-aborting DB errors (lock-wait / deadlock /
                # orphaned-savepoint) poison the shared session for every
                # subsequent person too — swallowing them here just yields
                # N identical 1305 errors and a wasted tick. Re-raise so the
                # handler restarts the whole idempotent tick on a fresh
                # session.
                if is_transaction_aborting_db_error(exc):
                    raise
                logger.exception(
                    "process_advance: failed for person %s", person.id,
                )
                summary["errors"].append({
                    "person_id": str(person.id),
                    "error": format_exception_for_log(exc, max_len=500),
                })

        # Identity Sync Center event hooks — commit the outer session
        # FIRST so the fresh session inside ``enqueue_person_sync`` can
        # observe the new ``employment_status='left'`` rows. The hook
        # is documented as never-raising, so a sync push failure here
        # cannot un-flip the persons we just transitioned.
        #
        # Fan out fire-and-forget via ``asyncio.create_task`` so a
        # bulk batch (50 leavers × 3 outbound targets at SCIM's 30 RPM
        # = ~5 min) doesn't serialize the entire process_advance run
        # behind downstream API rate limits. Each task internally
        # acquires the per-target Redis lock and serializes its own
        # work; the orchestrator is allowed to return as soon as the
        # tasks are scheduled. Lifecycle correctness (the
        # employment_status flip) is already committed by this point;
        # the event push is best-effort propagation that the daily
        # cron sweep would converge anyway.
        if transitioned_person_ids:
            await self.db.commit()
            from app.core.background import spawn
            from app.core.coordinators.leaver_cloud_sweep_coordinator import (
                sweep_then_offboard_sync,
            )

            for pid in transitioned_person_ids:
                # ``spawn`` retains a strong reference + logs failures so the
                # post-commit work can't be GC'd mid-flight.
                #
                # ``sweep_then_offboard_sync`` first deletes the leaver's
                # residual AWS Identity Center account assignments (the
                # untracked ones Helix has no grant row for), THEN fires the
                # offboard directory sync. Ordering is load-bearing: the
                # offboard sync's SCIM deprovision DELETEs the Identity Store
                # user, after which those assignments would become
                # un-attributable blank-name orphans (#aws-empty-name-records).
                spawn(
                    sweep_then_offboard_sync(pid),
                    name=f"offboard_sweep_sync:{pid}",
                )

        return summary

    # ── Orphaned-left recovery sweep ─────────────────────────

    async def recover_orphaned_left_processes(
        self,
        *,
        fallback_email: str | None = None,
    ) -> dict[str, Any]:
        """Self-heal offboarding processes that are still non-terminal
        (SCANNING / BLOCKED / READY) for a person who is ALREADY
        ``employment_status='left'``, driving each one to COMPLETED.

        Invariant being repaired: a ``left`` person must never have a
        non-terminal offboarding process. The normal advance pipeline
        (:meth:`run_tick`) only selects ``leaver_pending`` people — so
        once a person reaches ``left`` with a non-terminal process,
        NOTHING else owns finishing it. Three shapes occur:
          - BLOCKED + pending blockers: owner-bound infra never
            force-transferred (the original 2026-06-01 lifu.wu/chujie
            case).
          - READY: the dominant wedge — _advance_single flipped the
            person to 'left' and parked the process at READY because
            READY→COMPLETED (complete_process) had only a manual caller,
            so every autonomously-advanced leaver piled up in the
            "可关停 SSO" bucket (16 prod processes, 11 already
            SSO-suspended for weeks).
          - SCANNING: a flip-to-left that landed mid-scan; re-scanned
            here so the next tick can advance it.

        How a person reaches that state (prod incident 2026-06-01,
        lifu.wu / chujie): a flip-to-left whose transaction snapshot
        predated a concurrent :meth:`OffboardingService.start_process`
        blocker-INSERT commit read ZERO pending blockers and was
        allowed — the guard + listener pending reads are now
        ``with_for_update()`` current reads to close that window, but
        this sweep is the defence-in-depth that heals any orphan
        regardless of how it arose (HRIS direct write, manual DB edit,
        a future bypass).

        ``reconcile_all_active_processes`` does NOT help here: it only
        flips blockers whose linked provisioning task / transfer has
        already COMPLETED. Owner-bound infra (cloud_server / DNS /
        cloud_iam_user / email_group / hardware) is precisely what
        these orphans still hold, and only force-transfer reassigns it.

        Recovery runs the SAME pipeline :meth:`run_tick` uses, via
        ``_advance_single(..., bypass_grace=True)``: the 24h IT grace
        window is moot (the person is already gone), so each resource's
        owner is reassigned to the process's
        ``default_handover_person_id`` / manager fallback and the
        process is completed. ``_advance_single`` is idempotent — the
        ``flip_to_left`` at its tail re-asserts the guard and writes the
        same ``left`` value (the audit listener skips an unchanged
        transition), its orphan-seat / alert-cc sweeps are themselves
        idempotent, and force-transfer skips already-resolved blockers.

        Deliberately does NOT fire the Identity Sync Center event hook:
        the person was flipped to ``left`` (and pushed downstream) on
        the original transition; this sweep only repairs internal
        blocker / process state, so re-pushing would be redundant churn.
        """
        summary: dict[str, Any] = {
            "scanned": 0,
            "recovered": 0,
            "completed": 0,
            "rescanned": 0,
            "force_transferred_blockers": 0,
            "still_blocked": 0,
            "incidents_created": 0,
            "errors": [],
        }

        cfg_fallback = await get_platform_setting(
            "lifecycle.fallback_owner_email", default=None, session=self.db,
        )
        resolved_fallback = (
            (cfg_fallback or fallback_email or "").strip() or None
        )

        # Every person already ``employment_status='left'`` whose
        # offboarding process is NOT terminal (any of SCANNING / BLOCKED
        # / READY). ``run_tick`` only selects ``leaver_pending`` people,
        # so once a person reaches ``left`` with a non-terminal process,
        # NOTHING else owns driving it to COMPLETED:
        #   - BLOCKED + pending blockers: the original orphan case —
        #     owner-bound infra never force-transferred.
        #   - READY (with or without a stamped ``sso_suspended_at``): the
        #     dominant prod wedge (2026-06: 16 processes, 11 already
        #     SSO-suspended for weeks) — _advance_single previously
        #     flipped the person to 'left' and parked the process at
        #     READY because READY→COMPLETED had only a manual caller.
        #   - SCANNING: a flip-to-left that landed mid-scan; nothing
        #     re-scanned it, so it never reached BLOCKED/READY.
        # Earlier this method required ``status==BLOCKED`` AND a pending
        # blocker, so READY/SCANNING orphans (and BLOCKED-without-pending)
        # were structurally invisible to the one cron named to heal them.
        orphans = list((
            await self.db.execute(
                select(IdentityPerson)
                .options(selectinload(IdentityPerson.manager))
                .join(
                    OffboardingProcess,
                    OffboardingProcess.person_id == IdentityPerson.id,
                )
                .where(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status == "left",
                    OffboardingProcess.status.in_(
                        (PROC_SCANNING, PROC_BLOCKED, PROC_READY),
                    ),
                )
            )
        ).scalars().unique().all())

        summary["scanned"] = len(orphans)

        for person in orphans:
            # Per-person savepoint — one orphan's force-transfer failure
            # (missing fallback, connector error) must not abort the
            # others, mirroring run_tick's isolation.
            try:
                async with self.db.begin_nested():
                    # A person can have more than one OffboardingProcess
                    # row over their lifetime (re-hired then re-left). An
                    # unordered scalar() returns an arbitrary row and could
                    # heal the old terminal process while the live orphan
                    # stays stuck. Pick the most recent, matching the
                    # forensic/_detect_in_flight convention.
                    proc = await self.db.scalar(
                        select(OffboardingProcess)
                        .where(
                            OffboardingProcess.person_id == person.id,
                        )
                        .order_by(OffboardingProcess.created_at.desc())
                        .limit(1)
                    )
                    if proc is not None and proc.status == PROC_SCANNING:
                        # A SCANNING process has not finished enumerating
                        # blockers — feeding it to _advance_single (which
                        # reads only already-persisted blockers, it does
                        # NOT rescan) could complete it while resources
                        # are still unchecked. Re-run the scan instead;
                        # _recompute_status moves it to BLOCKED/READY and
                        # the NEXT tick advances it cleanly.
                        await self.offboarding.reconcile_blockers(
                            proc.id,
                            trace_id="recover_orphaned_left",
                            actor_user_id=None,
                        )
                        summary["rescanned"] += 1
                        continue
                    result = await self._advance_single(
                        person,
                        fallback_email=resolved_fallback,
                        bypass_grace=True,
                    )
                summary["force_transferred_blockers"] += result[
                    "force_transferred"
                ]
                summary["incidents_created"] += result["incidents_created"]
                if result["transitioned"]:
                    summary["recovered"] += 1
                    if result.get("completed"):
                        summary["completed"] += 1
                else:
                    # Force-transfer could not clear every blocker
                    # (e.g. missing fallback person → incident raised
                    # inside _advance_single). Leave it for the next
                    # tick + the compliance incident it already opened.
                    summary["still_blocked"] += 1
            except Exception as exc:  # noqa: BLE001
                # See run_tick: a transaction-aborting DB error poisons the
                # shared session for the remaining orphans, so re-raise for a
                # fresh-session retry instead of recording N cascaded 1305s.
                if is_transaction_aborting_db_error(exc):
                    raise
                logger.exception(
                    "recover_orphaned_left_processes: failed for person %s",
                    person.id,
                )
                summary["errors"].append({
                    "person_id": str(person.id),
                    "error": format_exception_for_log(exc, max_len=500),
                })

        # No Identity Sync Center fan-out here (see docstring): these
        # persons are already 'left' downstream.
        return summary

    # ── Per-person orchestration ─────────────────────────────

    async def _advance_single(
        self,
        person: IdentityPerson,
        *,
        fallback_email: str | None,
        bypass_grace: bool = False,
    ) -> dict[str, Any]:
        """Force-resolve all blockers, then atomically transition.

        Returns ``{force_transferred, transitioned, incidents_created,
        grace_held}``.

        ``bypass_grace=True`` (admin force-bypass path, 2026-06-02):
        ignore the T+0 24h IT-response window and treat the leaver
        like the 24h has already elapsed — physical force-transfer
        + revoke fire on this tick. Caller is responsible for the
        RBAC + audit gating; this kwarg is the mechanical plumbing.
        The grace card is also suppressed (its purpose is to give IT
        a heads-up so they can act manually during grace — moot once
        IT is the one explicitly skipping it). HRIS-late-detection
        is the canonical case: termination weeks ago, no T-N
        reminders fired, IT has drained resources manually, no
        operational reason to wait another 24h.
        """
        proc = await self.db.scalar(
            select(OffboardingProcess)
            .where(OffboardingProcess.person_id == person.id)
            # Most-recent process for a multi-offboarding person (re-hired
            # then re-left); an unordered scalar() could pick a stale
            # terminal row. Matches the forensic/_detect_in_flight order.
            .order_by(OffboardingProcess.created_at.desc())
            .limit(1)
        )
        if proc is None:
            # No process yet — the HR confirm path normally creates one.
            # Create on the fly so we don't silently skip a leaver.
            proc = await self.offboarding.start_process(
                person.id,
                person.termination_date or datetime.now(UTC).date(),
                trace_id="process_advance",
                actor_user_id=None,
            )

        assert proc is not None

        # Pre-launch P1 fix (Round 45 follow-up, CR-313): refuse to
        # advance / force-transfer when the process has been
        # terminally closed (PROC_CANCELLED / PROC_COMPLETED). Today
        # ``cancel_process`` restores ``employment_status='active'``
        # so the loop's leaver filter already prevents this — but
        # this is defense in depth against (a) future code paths
        # that cancel a process while leaving the person leaver_pending,
        # and (b) a manual DB edit by admin. Without this gate a
        # terminal process would have its blockers overwritten with
        # RESO_TRANSFERRED on the next tick, corrupting forensics.
        from app.domains.offboarding.models import (  # noqa: PLC0415
            PROC_CANCELLED,
            PROC_COMPLETED,
        )
        if proc.status in (PROC_CANCELLED, PROC_COMPLETED):
            logger.info(
                "process_advance: skipping person %s — process %s is "
                "terminal (%s)", person.id, proc.id, proc.status,
            )
            return {
                "force_transferred": 0,
                "transitioned": False,
                "incidents_created": 0,
                "grace_held": False,
                "skipped_reason": f"process_{proc.status}",
            }

        # Pre-termination local-day floor — HOISTED above ALL side effects
        # (2026-07-13 review P1). ``run_tick``'s selection loop already
        # defers any leaver whose ``local_today <= termination_date``, so it
        # never reaches this method before the last working day. But the
        # on-demand paths — ``advance_now`` / ``force_bypass_grace_advance``
        # (leaver_blocked/router.py) → ``_run_advance_for_process`` — call
        # ``_advance_single`` DIRECTLY with no date gate. The finalisation
        # floor further down (PR #440) only held the SSO-suspend/flip; the
        # T+0 work ABOVE it (identity-consumer PII-egress cut, immediate SSO/
        # VPN/cloud/AI revoke, owner-FK force-transfer under bypass) still
        # ran — so an admin clicking 立即推进/强制跳过宽限 on a leaver whose
        # last day is still in the FUTURE stripped a currently-working
        # employee's access days early (and, with bypass, rewrote their
        # asset/DNS/cloud ownership). A leaver is only actionable once their
        # own local last day has passed; refuse the whole advance until then.
        # ``bypass_grace`` collapses the POST-departure 24h IT grace, never
        # this PRE-departure floor. For run_tick this check is a no-op (its
        # people are always ``local_today > termination_date``).
        #
        # Boundary is STRICTLY-before (``<``), not ``<=``: acting ON the
        # local termination day (the person's last working day) via an
        # explicit admin bypass is deliberate, tested behaviour — force-
        # bypass expedites the software not_before and the T+0 identity-
        # consumer egress cut on the last day (test_force_bypass_expedites_
        # software / test_t0_unclaimed_disables_without_fallback). The
        # irreversible FINALISATION is still held on the last day by the
        # ``<=`` floor further down. What this hoisted floor forbids is the
        # genuinely-early case the review flagged: an advance for a leaver
        # whose last day is STILL IN THE FUTURE (days away), which stripped
        # a currently-working employee's access.
        if person.termination_date is not None:
            try:
                _floor_tz = resolve_timezone(person)
            except Exception:  # noqa: BLE001 — TZ resolution must never block
                _floor_tz = None
            _floor_today = (
                datetime.now(UTC).astimezone(_floor_tz).date()
                if _floor_tz is not None else datetime.now(UTC).date()
            )
            if _floor_today < person.termination_date:
                logger.info(
                    "process_advance: refusing on-demand advance for person "
                    "%s — local termination day (%s) is still in the future "
                    "(ref_today=%s); no revoke/transfer/finalisation run",
                    person.id, person.termination_date, _floor_today,
                )
                return {
                    "force_transferred": 0,
                    "transitioned": False,
                    "incidents_created": 0,
                    "grace_held": False,
                    "skipped_reason": "pre_termination_local_day",
                }

        # Load blockers — ``OffboardingProcess`` has no ``blockers``
        # relationship (see models.py); we query explicitly.
        blockers = list((
            await self.db.execute(
                select(OffboardingBlocker)
                .where(OffboardingBlocker.process_id == proc.id)
            )
        ).scalars().all())
        pending = [b for b in blockers if b.resolution == RESO_PENDING]
        force_count = 0
        incidents_created = 0

        # T+0 grace window (round-13 hardening, 2026-05-08; refined
        # round-58 2026-05-30): the day a process is first seen by the
        # worker we hold off on *physical* force-transfer for 24h and
        # push IT admins a heads-up. This matters most in HRIS-late-
        # detection scenarios (P01121: termination_date 2026-05-01 but
        # offboarding_process created 2026-05-08 — the leaver never
        # received T-7/T-3/T-1 reminders, so jumping straight to
        # manager-fallback gives nobody a chance to act).
        #
        # Round-58 (P0788, 2026-05-30): permission grants must NOT
        # participate in the 24h hold. They are revoked on the first
        # tick and the IT push lists only the remaining grace-eligible
        # work. See module-level
        # ``IMMEDIATE_REVOKE_RESOURCE_TYPES`` /
        # ``SCAN_SCHEDULED_RESOURCE_TYPES`` for the partition rationale.
        if pending:
            # T+0 PII-egress cut, BEFORE the fallback-gated buckets:
            # disable any unclaimed identity-consumer the leaver owned.
            # Disabling is a system action needing no successor, so — unlike
            # the owner-FK transfers below — it must NOT be held behind
            # ``_resolve_fallback_person`` (Step 1 / Step 3 both early-return
            # on a missing fallback) nor the 24h grace window. Without this,
            # a leaver with a dead manager chain and no IT-admin fallback
            # could leave a data tap live indefinitely. Resolved blockers are
            # dropped from ``pending`` so the immediate/grace buckets stay
            # byte-for-byte unchanged for every other resource type.
            consumer_pending = [
                b for b in pending
                if b.resource_type == RES_IDENTITY_CONSUMER_OWNER
            ]
            if consumer_pending:
                force_count += await self._disable_unclaimed_consumers(
                    person=person, process=proc, blockers=consumer_pending,
                )
                pending = [
                    b for b in pending
                    if b.resource_type != RES_IDENTITY_CONSUMER_OWNER
                ]
            immediate_revoke = [
                b for b in pending
                if b.resource_type in IMMEDIATE_REVOKE_RESOURCE_TYPES
            ]
            grace_eligible = [
                b for b in pending
                if b.resource_type not in IMMEDIATE_REVOKE_RESOURCE_TYPES
                and b.resource_type not in SCAN_SCHEDULED_RESOURCE_TYPES
            ]
            # ``SCAN_SCHEDULED_RESOURCE_TYPES`` (software_*) blockers
            # are intentionally absent from both buckets — their
            # resolution is owned by ProvisioningTask + not_before, not
            # the advance tick.

            fallback_person: IdentityPerson | None = None

            # Round-59 follow-up (2026-05-31, sharpened post-fix audit):
            # Pre-compute whether the IT grace card WILL fire in this
            # same tick so step 1's ``_force_transfer_all`` knows
            # whether to suppress its coalesced SSO bulk-revoke summary
            # (avoids duplicate listing when grace card already
            # enumerates immediate_revoked grants in its body).
            #
            # CRITICAL: ``now_utc`` MUST be computed ONCE and reused in
            # both the pre-check below AND the step-2 gate, otherwise a
            # leaver whose discovery falls in the [23.99h, 24.0h)
            # window pre-check can fall to >= 24h by the time step 2
            # runs (step 1 takes seconds to execute on a leaver with
            # many SSO grants × multi-channel notify). In that race
            # the pre-check says "grace card will fire, suppress
            # bulk-revoke", but step 2 actually skips the grace card
            # (>= 24h) — IT admins then receive ZERO summary of the
            # SSO revokes (per-system-owner cards still fire, but the
            # IT-pool signal is gone). Surfaced by the R59 post-fix
            # adversarial verify pass.
            #
            # Also: ``_notify_it_grace_window`` is idempotent per
            # (process_id, REMINDER_IT_GRACE) — on tick ≥ 2 it no-ops
            # via the already-logged check at ~line 786. If a new SSO
            # blocker materialises mid-grace-window (HRIS resync /
            # fresh approval), tick 2 would suppress bulk-revoke AND
            # the grace card would no-op — same blind spot. Gate also
            # on the idempotency status so bulk-revoke fires when the
            # grace card already burnt its slot.
            now_utc = datetime.now(UTC)
            discovery_dt = (
                proc.created_at if proc.created_at is not None else now_utc
            )
            now_naive = now_utc.replace(tzinfo=None)
            disc_naive = (
                discovery_dt.replace(tzinfo=None)
                if discovery_dt.tzinfo is not None else discovery_dt
            )
            hours_since_discovery = (
                (now_naive - disc_naive).total_seconds() / 3600.0
            )
            grace_card_already_burnt = await self._grace_card_already_logged(
                proc.id,
            )
            # Round-61 P0 fix: split the time-window HOLD decision from
            # the card-emission decision. ``grace_window_open`` is the
            # ONLY thing that may hold the force-transfer back — it must
            # depend on the 24h clock alone, NOT on whether the grace
            # card was already logged. Before this fix the hold reused
            # ``grace_card_will_fire`` (which includes ``not already_burnt``),
            # so on tick 2 (cron every 15 min) the already-burnt card
            # made ``grace_card_will_fire`` False, Step 2 fell through to
            # Step 3, and ``_force_transfer_all`` overwrote the asset /
            # network owner FKs only ~15 min after discovery — collapsing
            # the documented 24h IT grace window (P01121 HRIS-late-
            # detection protection) to a single tick.
            # 2026-06-02 admin force-bypass: when the caller passes
            # ``bypass_grace=True``, the 24h hold is collapsed to zero
            # regardless of ``hours_since_discovery``. The grace card
            # is also suppressed (see ``grace_card_will_fire`` below).
            grace_window_open = (
                bool(grace_eligible)
                and hours_since_discovery < 24
                and not bypass_grace
            )
            # ``grace_card_will_fire`` stays gated on the idempotency
            # status — it drives (a) whether ``_notify_it_grace_window``
            # actually emits the card this tick and (b) the Step-1
            # bulk-revoke-summary suppression (so the SSO bulk-revoke
            # card still fires when the grace card already burnt its slot).
            grace_card_will_fire = (
                grace_window_open
                and not grace_card_already_burnt
            )

            # ── Step 1: immediate revoke (no grace) ──
            if immediate_revoke:
                fallback_person = await self._resolve_fallback_person(
                    person, fallback_email=fallback_email,
                )
                if fallback_person is None:
                    _, created = await self.incidents.create_or_upsert_incident(
                        incident_type=TYPE_MANAGER_MISSING,
                        severity=SEVERITY_CRITICAL,
                        person_id=person.id,
                        # Dedup per offboarding PROCESS, not just per
                        # incident_type. ``create_or_upsert_incident``
                        # only adds resource_type/resource_id to its
                        # dedup WHERE when non-None; without them the
                        # lookup matches ANY open incident of this type
                        # regardless of person, so a second stuck leaver
                        # in the same batch would overwrite the first's
                        # title/detail and the admin would never see it
                        # (forensics loss). ``str(proc.id)`` not the raw
                        # UUID — resource_id is CHAR(36) and asyncmy
                        # binds a uuid.UUID as bytes → silent 0-row match.
                        resource_type="offboarding_process",
                        resource_id=str(proc.id),
                        resource_label=person.display_name or person.work_email,
                        title=(
                            f"{person.display_name or person.work_email} 无法自动 revoke 权限 — "
                            "manager 链与 IT Admin 兜底人都不可用"
                        ),
                        detail={
                            "termination_date": (
                                person.termination_date.isoformat()
                                if person.termination_date else None
                            ),
                            "immediate_revoke_blocker_count": len(immediate_revoke),
                        },
                    )
                    if created:
                        incidents_created += 1
                    return {
                        "force_transferred": 0,
                        "transitioned": False,
                        "incidents_created": incidents_created,
                        "grace_held": False,
                        "hold_reason": "missing_fallback",
                    }
                force_count += await self._force_transfer_all(
                    person=person,
                    process=proc,
                    blockers=immediate_revoke,
                    new_owner=fallback_person,
                    # Round-59 follow-up: suppress bulk-revoke admin
                    # summary when the grace card will fire in the same
                    # tick — the grace card already lists immediate-
                    # revoked grants in its body.
                    emit_sso_bulk_revoke_summary=not grace_card_will_fire,
                )
                # Round-58 follow-up (logic scan F1): ``_force_transfer_all``
                # per-blocker SAVEPOINT-isolates failures, so a row whose
                # revoke raised (e.g. Round-21 RBAC FORBIDDEN against the
                # cloud-side grant API) stays ``RESO_PENDING``. Re-filter
                # so the IT grace card + reminder-log audit only attribute
                # rows that ACTUALLY resolved this tick — pre-filter the
                # original input list would have overstated success and
                # ``immediate_revoked_blocker_ids`` would have lied about
                # which grants are now revoked vs still active.
                immediate_revoke = [
                    b for b in immediate_revoke
                    if b.resolution != RESO_PENDING
                ]

            # ── Step 2: grace decision (physical transfer types only) ──
            # ``hours_since_discovery`` is the SAME value computed
            # before step 1 (above) — Round-59 sharpened-audit fix
            # eliminates the boundary race where a second clock-read
            # could land >= 24h after step 1's elapsed work pushed past
            # the threshold. Round-61 P0 fix: the HOLD is gated on
            # ``grace_window_open`` (24h clock only). The grace card is
            # emitted only on ``grace_card_will_fire`` (first tick within
            # the window); ``_notify_it_grace_window`` is itself
            # idempotent per (process_id, REMINDER_IT_GRACE), so tick ≥ 2
            # would no-op the card anyway — but the process MUST still
            # hold until the 24h window genuinely expires.
            if grace_eligible:
                if grace_window_open:
                    if grace_card_will_fire:
                        await self._notify_it_grace_window(
                            proc=proc,
                            person=person,
                            pending=grace_eligible,
                            immediate_revoked=immediate_revoke,
                        )
                    return {
                        "force_transferred": force_count,
                        "transitioned": False,
                        "incidents_created": incidents_created,
                        "grace_held": True,
                        # Round-22 contract: every non-transitioned
                        # return path declares ``hold_reason`` explicitly
                        # so the router / cron summary dispatcher can't
                        # silently fall through to a wrong label.
                        "hold_reason": "grace",
                    }

                # ── Step 3: post-grace force-transfer ──
                if fallback_person is None:
                    fallback_person = await self._resolve_fallback_person(
                        person, fallback_email=fallback_email,
                    )
                    if fallback_person is None:
                        _, created = await self.incidents.create_or_upsert_incident(
                            incident_type=TYPE_MANAGER_MISSING,
                            severity=SEVERITY_CRITICAL,
                            person_id=person.id,
                            # Per-process dedup key — see step-1 comment.
                            # Same incident_type for the same leaver
                            # (step-1 immediate-revoke ↔ step-3 force-
                            # transfer) intentionally collapses to one
                            # "no fallback person" ticket per process.
                            resource_type="offboarding_process",
                            resource_id=str(proc.id),
                            resource_label=person.display_name or person.work_email,
                            title=(
                                f"{person.display_name or person.work_email} 无法自动转让 — "
                                "manager 链与 IT Admin 兜底人都不可用"
                            ),
                            detail={
                                "termination_date": (
                                    person.termination_date.isoformat()
                                    if person.termination_date else None
                                ),
                                "pending_blocker_count": len(grace_eligible),
                            },
                        )
                        if created:
                            incidents_created += 1
                        return {
                            "force_transferred": force_count,
                            "transitioned": False,
                            "incidents_created": incidents_created,
                            "grace_held": False,
                            "hold_reason": "missing_fallback",
                        }
                force_count += await self._force_transfer_all(
                    person=person,
                    process=proc,
                    blockers=grace_eligible,
                    new_owner=fallback_person,
                )

        # Transition — the caller wraps the whole call in begin_nested
        # so force_transfer + this block commit atomically (or both
        # roll back). No inner savepoint needed here.
        refreshed_blockers = list((
            await self.db.execute(
                select(OffboardingBlocker)
                .where(OffboardingBlocker.process_id == proc.id)
            )
        ).scalars().all())
        # Round-58 (2026-05-30) — split the still-pending bucket:
        #
        #   - ``unhandled_pending`` is a force-transfer-eligible row
        #     that _force_transfer_all should have resolved but didn't.
        #     This is a real bug condition → LEAVER_ORPHAN incident.
        #
        #   - ``scheduled_pending`` is a SCAN_SCHEDULED_RESOURCE_TYPES
        #     row (software_seat / permission / account) whose
        #     resolution is owned by the ProvisioningTask pipeline.
        #     Pre-58 lumped these into ``still_pending``, so a leaver
        #     with an in-flight ``not_before`` revoke task triggered a
        #     spurious LEAVER_ORPHAN incident every 15 minutes
        #     (upsert idempotent on the row, but with the wrong type +
        #     wrong title, mis-routing IT attention).
        unhandled_pending = [
            b for b in refreshed_blockers
            if b.resolution == RESO_PENDING
            and b.resource_type not in SCAN_SCHEDULED_RESOURCE_TYPES
        ]
        scheduled_pending = [
            b for b in refreshed_blockers
            if b.resolution == RESO_PENDING
            and b.resource_type in SCAN_SCHEDULED_RESOURCE_TYPES
        ]
        if unhandled_pending:
            # Shouldn't happen if force_transfer_all ran cleanly;
            # treat as blocker and raise an incident. Next tick retries.
            _, created = await self.incidents.create_or_upsert_incident(
                incident_type=TYPE_LEAVER_ORPHAN,
                severity=SEVERITY_CRITICAL,
                person_id=person.id,
                # Per-process dedup key — see step-1 comment. Without
                # it two stuck leavers collapse into one orphan ticket
                # and one's blocker_ids overwrite the other's.
                resource_type="offboarding_process",
                resource_id=str(proc.id),
                resource_label=person.display_name or person.work_email,
                title=(
                    f"{person.display_name or person.work_email} 仍有 "
                    f"{len(unhandled_pending)} 项 blocker 无法强制转让"
                ),
                detail={
                    "blocker_ids": [
                        str(b.id) for b in unhandled_pending
                    ],
                },
            )
            if created:
                incidents_created += 1
            return {
                "force_transferred": force_count,
                "transitioned": False,
                "incidents_created": incidents_created,
                "grace_held": False,
                "hold_reason": "pending_blockers",
                "pending_blocker_count": len(unhandled_pending),
            }
        if scheduled_pending:
            # 2026-06-09 admin force-bypass: software_* revokes are
            # ProvisioningTasks scheduled at ``not_before = leaver_date +
            # 1 (local midnight)`` — the normal advance tick respects
            # that schedule. But ``bypass_grace=True`` means "this person
            # is gone, stop waiting" — so collapse their ``not_before``
            # to NOW. The next ``provisioning.dispatch`` tick (~100s)
            # then fires auto_api revokes immediately and pings the
            # manual_notify owners right away, instead of parking the
            # whole offboarding until tonight. Idempotent: only pulls
            # rows still PENDING with a future ``not_before``.
            if bypass_grace:
                await self._expedite_software_revokes(
                    scheduled_pending, person=person, proc=proc,
                )
            # ProvisioningTask pipeline still working. Leave proc in
            # BLOCKED — the task's success callback will flip the
            # blocker to RESO_REVOKED_AUTO and the next advance tick
            # (after that callback) will find pending == 0 and
            # transition cleanly.
            return {
                "force_transferred": force_count,
                "transitioned": False,
                "incidents_created": incidents_created,
                "grace_held": False,
                "hold_reason": "awaiting_provisioning",
                "pending_blocker_count": len(scheduled_pending),
            }

        # Local-last-day floor (PR #440 invariant) — DEFENSE IN DEPTH. As of
        # the 2026-07-13 review fix the primary floor is HOISTED to the top of
        # this method (above every side effect), so a pre-termination leaver
        # returns long before here. This second copy is retained as a belt-
        # and-suspenders guard on the irreversible FINALISATION (PROC_READY →
        # suspend_sso → flip_to_left) in case a future refactor moves or
        # weakens the hoisted check; it must never be the ONLY floor (the T+0
        # revoke/transfer above it would then fire on a still-working future
        # leaver — the exact bug the hoist fixed). ``bypass_grace`` collapses
        # the POST-departure 24h IT grace, never this PRE-departure floor.
        if person.termination_date is not None:
            try:
                _tz = resolve_timezone(person)
            except Exception:  # noqa: BLE001 — TZ resolution must never block
                _tz = None
            _ref_today = (
                datetime.now(UTC).astimezone(_tz).date() if _tz is not None
                else datetime.now(UTC).date()
            )
            if _ref_today <= person.termination_date:
                logger.info(
                    "process_advance: holding FINALISATION for person %s — "
                    "still within local termination day (%s, ref_today=%s)",
                    person.id, person.termination_date, _ref_today,
                )
                return {
                    "force_transferred": force_count,
                    "transitioned": False,
                    "incidents_created": incidents_created,
                    "grace_held": False,
                    "skipped_reason": "pre_termination_local_day",
                }

        proc.status = PROC_READY
        proc.ready_at = proc.ready_at or datetime.now(UTC)
        # Flush the status change before calling suspend_sso. The
        # suspend_sso implementation calls _get_process(lock=True) which
        # does a fresh SELECT FOR UPDATE from the DB — without flushing here
        # it reads the pre-READY status and raises PROCESS_NOT_READY.
        await self.db.flush()

        # Dispatch any pending revoke tasks (auto_revoke blockers
        # that created provisioning tasks). suspend_sso is idempotent.
        await self.offboarding.suspend_sso(
            proc.id, trace_id="process_advance", actor_user_id=None,
        )

        # Final flip — this is the invariant-critical line. Routed
        # through LeaverBlockGuard so the audit / listener layers
        # observe a guarded write (no out-of-band warning) and any
        # newly-appeared blocker between force-transfer flush and
        # this point is caught with a structured rejection rather
        # than a silent listener trap.
        await self.leaver_guard.flip_to_left(
            person,
            via=LEAVER_VIA_PROCESS_ADVANCE,
            trace_id="process_advance",
            actor_user_id=None,
            last_synced_at=datetime.now(UTC),
        )

        # ── Terminal transition: auto-complete the process ──
        # Before 2026-06, _advance_single stopped after flip_to_left:
        # the person was 'left' but the OffboardingProcess was parked at
        # PROC_READY forever, because READY→COMPLETED (complete_process)
        # had exactly ONE caller — the manual admin "Complete" button. No
        # cron owned the terminal transition, so every autonomously
        # advanced leaver piled up in the "可关停 SSO" (ready) bucket and
        # never closed (prod: 16 processes wedged, 11 with SSO suspended
        # for weeks). Now that suspend_sso + flip_to_left have run, drive
        # the process to COMPLETED in this same atomic tick.
        #
        # Guarded: complete_process re-asserts its own invariants
        # (sso_suspended_at set, person 'left', not BLOCKED/SCANNING). If
        # a revoke task failed and suspend_sso therefore did NOT stamp
        # sso_suspended_at, complete_process raises PROCESS_NOT_READY — we
        # swallow it and leave the process at READY for the next tick /
        # admin to resolve the failed task. An AppException here must not
        # abort the invariant-critical flip_to_left above.
        completed = False
        await self.db.flush()
        try:
            done = await self.offboarding.complete_process(
                proc.id, trace_id="process_advance", actor_user_id=None,
            )
            completed = done.status == PROC_COMPLETED
        except AppException as exc:
            logger.info(
                "process_advance: process %s flipped to left + SSO "
                "dispatched but not auto-completed (%s); left at READY "
                "for next tick / manual completion",
                proc.id, getattr(exc, "code", type(exc).__name__),
            )

        # Review 2026-07-08 fix (#1739): replay the DEFERRED identity-plane
        # session/account cutoff. A future-dated leaver's
        # ``dispatch_post_commit_actions`` returned early
        # (``deferred_until_leaver_date``) at process-start and left
        # ``session_enforcement`` NULL, trusting "T+0 will call it again" —
        # but nothing did, so the leaver's Google ``users.signOut`` + account
        # suspend never fired and the NULL ledger hid it from the
        # ``offboarding_session_not_cut`` scan. We are past the
        # pre-termination local-day floor above, so leaver_date has arrived:
        # this is the correct moment to catch up. The call is a no-op when a
        # ledger already exists (event-path leavers who were cut at start),
        # so it is idempotent and cheap to run every finalisation. A cutoff
        # FAILURE is recorded in the ledger (not raised) by
        # _push_force_logout and then surfaced by the data-quality scan +
        # the manual reenforce retry — it must not abort this invariant-
        # critical tick, hence the defensive try/except.
        async with best_effort(
            self.db, label=f"advance: deferred session cutoff catchup for {proc.id}", log=logger,
        ) as catchup_step:
            replayed = await self.offboarding.catchup_deferred_session_cutoff(
                proc.id, trace_id="process_advance",
            )
            if replayed:
                await self.db.flush()
        if not catchup_step.ok:
            logger.error(
                "process_advance: deferred session-cutoff catch-up failed "
                "for process %s — force-logout may not have replayed; the "
                "offboarding_session_not_cut scan + manual re-enforce cover "
                "the residual",
                proc.id,
            )

        # Identity Sync Center event hook is fired by the OUTER
        # ``run_tick`` loop AFTER the session commits — see the
        # comment block there for why. Firing it here would race
        # the unflushed ``employment_status='left'`` write: the hook
        # opens its own session via ``async_session_factory()`` and
        # would read the pre-flip value, sending a current-state
        # upsert to Navan / SCIM at the worst possible moment.

        return {
            "force_transferred": force_count,
            "transitioned": True,
            "completed": completed,
            "incidents_created": incidents_created,
            "grace_held": False,
            # Round-22 contract: success path declares ``hold_reason``
            # = None so the router / cron summary dispatcher reads the
            # ``transitioned=True`` branch first and never falls
            # through to the hold_reason switch. Explicit > implicit.
            "hold_reason": None,
        }

    async def _expedite_software_revokes(
        self,
        scheduled_pending: list[OffboardingBlocker],
        *,
        person: IdentityPerson,
        proc: OffboardingProcess,
    ) -> None:
        """Force-bypass only: pull the ``not_before`` of the software_*
        revoke ProvisioningTasks forward to NOW so the dispatcher fires
        them on its next tick instead of waiting until the leaver's
        local-midnight schedule.

        Only touches tasks still ``status='pending'`` with a future
        ``not_before`` — already-running / completed / past-due rows are
        left alone (idempotent, safe to re-click). The actual revoke
        (auto_api) or owner-ping (manual_notify) is still done by the
        ``provisioning.dispatch`` cron — we only un-gate the schedule,
        never call a connector inline.
        """
        from sqlalchemy import update  # noqa: PLC0415

        from app.audit.service import AuditService  # noqa: PLC0415
        from app.domains.provisioning.models import (  # noqa: PLC0415
            STATUS_PENDING,
            ProvisioningTask,
        )

        task_ids = [
            b.provisioning_task_id
            for b in scheduled_pending
            if b.provisioning_task_id is not None
        ]
        if not task_ids:
            return
        now = datetime.now(UTC)
        result = await self.db.execute(
            update(ProvisioningTask)
            .where(
                ProvisioningTask.id.in_(task_ids),
                ProvisioningTask.status == STATUS_PENDING,
                ProvisioningTask.not_before.is_not(None),
                ProvisioningTask.not_before > now,
            )
            .values(not_before=now, updated_at=now)
        )
        expedited = result.rowcount or 0
        if expedited:
            logger.info(
                "force-bypass: expedited %d software-revoke task(s) for "
                "person %s (process %s) — not_before pulled to now",
                expedited, person.id, proc.id,
            )
            await AuditService(self.db).log(
                trace_id="process_advance",
                actor_type="system",
                actor_user_id=None,
                action="provisioning.not_before_expedited",
                status="success",
                resource_type="offboarding_process",
                resource_id=str(proc.id),
                detail={
                    "person_id": str(person.id),
                    "expedited_task_count": expedited,
                    "task_ids": [str(t) for t in task_ids],
                    "reason": "force_bypass_grace",
                },
            )

    # ── T+0 grace-window IT push ─────────────────────────────

    async def _grace_card_already_logged(self, process_id: UUID) -> bool:
        """Return True iff ``REMINDER_IT_GRACE`` was already stamped for
        this process (i.e., ``_notify_it_grace_window`` will no-op on
        this tick).

        Used by ``_advance_single`` Round-59 sharpened-audit fix to
        decide whether the SSO bulk-revoke admin summary card must
        fire — if grace card slot is already burnt, the grace card
        won't fire AND the bulk-revoke card is the only IT-admin
        signal about the SSO revokes happening this tick.
        """
        from app.domains.offboarding.models import (  # noqa: PLC0415
            REMINDER_IT_GRACE,
            OffboardingReminderLog,
        )

        row = await self.db.scalar(
            select(OffboardingReminderLog.id).where(
                OffboardingReminderLog.process_id == process_id,
                OffboardingReminderLog.reminder_kind == REMINDER_IT_GRACE,
            )
        )
        return row is not None

    async def _notify_it_grace_window(
        self,
        *,
        proc: OffboardingProcess,
        person: IdentityPerson,
        pending: list[OffboardingBlocker],
        immediate_revoked: list[OffboardingBlocker] | None = None,
    ) -> None:
        """One-time IT-admin push during the T+0 grace window.

        Idempotent via ``offboarding_reminder_log`` — the second tick
        within the same 24h grace window is a no-op. Channels:
        feishu + email to every active global_admin / it_admin. Body
        names each pending blocker (physical transfer types) so IT can
        decide who inherits before tomorrow's force-transfer fires.

        ``immediate_revoked`` (Round-58, 2026-05-30) lists the
        permission grants that were resolved on this same tick — these
        are surfaced in the push body so IT knows what already happened
        and doesn't waste time looking for them. They are NEVER part of
        the "24h pending" count.
        """
        from app.domains.offboarding.models import (  # noqa: PLC0415
            REMINDER_IT_GRACE,
            OffboardingReminderLog,
        )

        # Idempotency check — already pushed within this grace window?
        already = await self.db.scalar(
            select(OffboardingReminderLog.id).where(
                OffboardingReminderLog.process_id == proc.id,
                OffboardingReminderLog.reminder_kind == REMINDER_IT_GRACE,
            )
        )
        if already is not None:
            return

        # Resolve IT admin recipients.
        from app.auth.models import AuthRole, AuthUserRole  # noqa: PLC0415
        admin_ids = list((
            await self.db.execute(
                select(AuthUserRole.user_id)
                .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
                .where(AuthRole.name.in_(("global_admin", "it_admin")))
                # Dedup: an admin holding both global_admin and the legacy
                # it_admin alias would otherwise be notified twice.
                .distinct()
            )
        ).scalars().all())
        admins: list[IdentityPerson] = []
        for aid in admin_ids:
            admin = await self.db.get(IdentityPerson, aid)
            if (
                admin is not None
                and admin.employment_status == "active"
                and admin.deleted_at is None
            ):
                admins.append(admin)
        if not admins:
            logger.warning(
                "T+0 grace push for %s skipped: no active IT admins",
                person.id,
            )
            return

        # Pretty blocker list.
        blocker_lines = "\n".join(
            f"  · [{b.resource_type}] {b.label}" for b in pending
        )
        # Round-58: surface the permission grants we resolved on this
        # same tick so IT has the full picture (already-done + pending).
        # When empty (no immediate revokes this tick) we emit an empty
        # ``immediate_revoked_block`` so the template renders nothing —
        # avoids the awkward "本次已立即回收 (0 项): " empty section.
        immediate_revoked = immediate_revoked or []
        if immediate_revoked:
            immediate_revoked_lines = "\n".join(
                f"  · [{b.resource_type}] {b.label}" for b in immediate_revoked
            )
            immediate_revoked_block_zh = (
                f"本次已立即回收 ({len(immediate_revoked)} 项,权限类无需等待):\n"
                f"{immediate_revoked_lines}\n\n"
            )
            immediate_revoked_block_en = (
                f"Already revoked this tick ({len(immediate_revoked)} "
                f"permission grant(s)):\n{immediate_revoked_lines}\n\n"
            )
            immediate_revoked_block_ja = (
                f"今回のチェックで即時回収済み(権限 {len(immediate_revoked)} 件、"
                f"待機不要):\n{immediate_revoked_lines}\n\n"
            )
        else:
            immediate_revoked_lines = ""
            immediate_revoked_block_zh = ""
            immediate_revoked_block_en = ""
            immediate_revoked_block_ja = ""
        portal_url = await get_platform_setting(
            "app.frontend_base_url",
            default="(未配置门户地址)",
            session=self.db,
        )
        # Title goes through the template; we still pass it for the
        # subject_text / e-mail subject path - the template registry
        # carries subject + subject_en + body_text + body_text_en and
        # the dispatcher fills both. Body_text="(see template)" tells
        # the dispatcher to render from the template, which has NO
        # inline URL - the CTA button (url=) is the only clickable
        # element so dev / staging never leak ``localhost`` here.
        person_label = person.display_name or person.work_email
        ctx = {
            "person_name": person_label,
            "person_email": person.work_email,
            "employee_no": person.employee_no or "-",
            "leaver_date": proc.leaver_date.isoformat(),
            "discovery_date": proc.created_at.date().isoformat(),
            "pending_count": len(pending),
            "blocker_lines": blocker_lines,
            "immediate_revoked_count": len(immediate_revoked),
            "immediate_revoked_lines": immediate_revoked_lines,
            # Pre-formatted "已立即回收" block — empty string when no
            # immediate revokes happened this tick so the section
            # renders nothing. Locale-fragment triple: each language's
            # template interpolates its own variant.
            "immediate_revoked_block_zh": immediate_revoked_block_zh,
            "immediate_revoked_block_en": immediate_revoked_block_en,
            "immediate_revoked_block_ja": immediate_revoked_block_ja,
            "portal_url": (
                f"{portal_url}/admin/lifecycle/person/{person.id}"
            ),
        }

        # ENQUEUE (outbox-only), NOT dispatch. ``run_tick`` calls this from
        # inside its per-person ``begin_nested()`` savepoint, and the outer
        # transaction commits only once at the end of the whole leaver loop.
        # dispatch()'s synchronous Feishu+email HTTP would run inside that
        # long-lived transaction — and each outbox INSERT FK-locks the
        # recipient IT-admin's ``identity_persons`` row (notification_outbox
        # .recipient_person_id FK). On a bulk-leaver day (layoff / first tick
        # after a redeploy backlog) that holds shared FK locks on every active
        # admin's person row across N_admins×2 external channel sends until
        # the loop's commit, blocking those admins' login ``UPDATE`` on
        # identity_persons (MySQL 1205) — the exact #1837 login-outage class.
        # Enqueue durable rows now (DB-only, no external HTTP in the txn); the
        # hourly ``notification.outbox_retry`` worker delivers them. A 24h
        # grace reminder tolerates that latency; a login outage does not.
        enqueued_channels: set[str] = set()
        for admin in admins:
            try:
                await self.notify.enqueue(
                    NotificationEnvelope(
                        recipient=admin,
                        title=(
                            f"[离职宽限期] {person_label} "
                            f"({person.employee_no or '-'}) 仍有 "
                            f"{len(pending)} 项资源未处理 - 24h 后系统将兜底"
                        ),
                        body_text="(see template)",
                        channels=["feishu", "email"],
                        template_key="leaver_it_grace",
                        context=ctx,
                        url=(
                            f"{portal_url}/admin/lifecycle/person/{person.id}"
                        ),
                        related_resource_type="offboarding_process",
                        related_resource_id=proc.id,
                    ),
                )
                enqueued_channels.update(("feishu", "email"))
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "leaver_it_grace enqueue to %s failed: %s",
                    admin.work_email, exc,
                )

        # Log the reminder so the next tick is a no-op. Records the channels
        # we durably ENQUEUED (delivery is done post-commit by the outbox
        # retry worker) — this marker's contract is "did we fire this
        # reminder", not "did the SMTP/Feishu call return 200".
        self.db.add(OffboardingReminderLog(
            process_id=proc.id,
            reminder_kind=REMINDER_IT_GRACE,
            channels=sorted(enqueued_channels),
            detail={
                "admin_ids": [str(a.id) for a in admins],
                "blocker_ids": [str(b.id) for b in pending],
                "blocker_count": len(pending),
                # Round-58: forensic trail of which permission grants
                # were already revoked on this same tick — surfaces in
                # the audit log so reconcilers can see "the grace card
                # said 3 pending, but the system already revoked 8
                # permissions in the same minute". Without this, a
                # later "why did access end at T+0?" review can't tie
                # the revoke back to the grace tick.
                "immediate_revoked_blocker_ids": [
                    str(b.id) for b in immediate_revoked
                ],
                "immediate_revoked_count": len(immediate_revoked),
            },
        ))
        await self.db.flush()

    # ── Fallback resolution (manager chain → IT Admin → settings) ─

    async def _resolve_fallback_person(
        self,
        person: IdentityPerson,
        *,
        fallback_email: str | None,
    ) -> IdentityPerson | None:
        """Resolve the gate / default recipient for the leaver.

        Returns the first ``active`` candidate in:
          1. Manager chain — direct manager, then skip-level, … until
             an active, not-soft-deleted person is found.
          2. IT admin pool — any active ``global_admin`` / ``it_admin``.
          3. Configured fallback email (``OFFBOARDING_FALLBACK_EMAIL``).

        Two roles in the T+0 tick:
          - **Gate** (step 1, immediate-revoke; step 3, force-transfer):
            if this returns ``None`` the process holds and a compliance
            incident is opened — there is literally no one to attribute
            the action to.
          - **Default recipient** (step 3): passed as ``default_new_owner``
            to ``_force_transfer_all`` → ``_handle_one_blocker``. For
            hardware the per-asset office admin is preferred via
            ``_resolve_hardware_recipient``; for non-hardware the IT-set
            ``OffboardingProcess.default_handover_person_id`` short-circuits
            inside ``_handle_one_blocker``. This function intentionally
            does NOT honour the handover person itself — keeping the gate
            decision orthogonal from the per-blocker recipient lets a
            misconfigured handover person fail one blocker without
            breaking the rest of the tick.
        """
        visited: set[UUID] = {person.id}
        candidate: IdentityPerson | None = None
        # Climb the manager chain
        current = person
        while current.manager_person_id and current.manager_person_id not in visited:
            visited.add(current.manager_person_id)
            mgr = await self.db.get(IdentityPerson, current.manager_person_id)
            if mgr is None:
                break
            if (
                mgr.employment_status == "active"
                and mgr.deleted_at is None
            ):
                candidate = mgr
                break
            current = mgr
        if candidate is not None:
            return candidate

        # IT admin pool — any active global admin
        from app.auth.models import AuthRole, AuthUserRole
        admin_ids = list((
            await self.db.execute(
                select(AuthUserRole.user_id)
                .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
                .where(AuthRole.name.in_(("global_admin", "it_admin")))
            )
        ).scalars().all())
        for admin_id in admin_ids:
            admin = await self.db.get(IdentityPerson, admin_id)
            if (
                admin is not None
                and admin.employment_status == "active"
                and admin.deleted_at is None
            ):
                return admin

        # Configured fallback email
        if fallback_email:
            fb = await self.db.scalar(
                select(IdentityPerson).where(
                    IdentityPerson.work_email == fallback_email,
                    IdentityPerson.deleted_at.is_(None),
                )
            )
            if fb is not None and fb.employment_status == "active":
                return fb
        return None

    # ── Force-transfer (the actual FK rewrite) ──────────────

    async def _resolve_hardware_recipient(
        self,
        *,
        asset: AssetItem | None,
        default: IdentityPerson,
    ) -> IdentityPerson:
        """Pick the new owner for a hardware blocker.

        The device goes to the administrator of the location it sits
        at. Cross-geo offboarding (a Sunnyvale leaver hands their
        MacBook to Kristine Ye, not to Beijing manager Jason Meng) is
        the dominant real-world flow, so that administrator is a far
        better default than the manager.

        This used to match ``asset.location`` STRICTLY against
        ``asset_office_addresses.city``, which held only for the six
        Chinese cities: an asset at ``美国`` never equalled the office
        city ``Sunnyvale``, so every US, Singapore, Japan and Malaysia
        device — 300 of 1099 in production, 27% of the fleet —
        silently fell through to the manager, the exact outcome the
        strict match was meant to prevent. Resolution now goes through
        the managed-location master, the same resolver every other
        location string in the asset domain uses, so 美国 → 美国 and
        the administrator is found.

        Falls back to the per-leaver default (manager → IT admin →
        configured fallback email) when:
          - the asset has no location, or
          - the location resolves to no managed row, or
          - that location has no single accountable administrator.
        """
        if asset is None or not asset.location:
            return default
        location_id = await resolve_location_id_by_name(self.db, asset.location)
        if location_id is None:
            return default
        primaries = await resolve_location_primary_admins_bulk(
            self.db, [location_id],
        )
        admin_id = primaries.get(location_id)
        if admin_id is None:
            return default
        admin = await self.db.get(IdentityPerson, admin_id)
        if admin is None or admin.deleted_at is not None:
            return default
        if admin.employment_status != "active":
            return default
        return admin

    def _resolve_hardware_disposition(
        self,
        *,
        process: OffboardingProcess,
        asset: AssetItem,
        fallback_recipient: IdentityPerson,
    ) -> dict[str, Any]:
        """Resolve the disposition for one hardware blocker.

        Reads ``process.hardware_disposition[str(asset.id)]`` (set by
        the offboarding console). When no entry exists for this asset,
        defaults to ``keep`` (held_by_leaver) — the user instruction
        (2026-05-20): leaver retains the device when no one acted, IT
        receives a same-day notification, no other process blocks.
        ``_default_fallback`` marks "nobody chose this, the default
        kicked in" so the downstream IT notification can distinguish
        intentional ``keep`` from auto-fallback ``keep``.

        Returns a dict shape compatible with ``HardwareDispositionEntry``:
          ``{"action": "...", "office_id": "...|None", "person_id": "...|None"}``
        ``office_id`` / ``person_id`` are strings (JSON-shape) — apply
        converts to UUID at the point of use.
        """
        raw = process.hardware_disposition or {}
        entry = raw.get(str(asset.id))
        if isinstance(entry, dict) and entry.get("action"):
            return entry
        return {
            "action": "keep",
            "office_id": None,
            "person_id": None,
            "_default_fallback": True,
        }

    async def _apply_hardware_disposition(
        self,
        *,
        blocker: OffboardingBlocker,
        asset: AssetItem,
        person: IdentityPerson,
        disposition: dict[str, Any],
        fallback_recipient: IdentityPerson,
        prev_assigned_to,
        was_borrower: bool,
    ) -> tuple[bool, dict[str, Any] | None]:
        """Execute one hardware disposition on a single asset.

        Audit / notification attribution policy (PR-#406 review fix):
          * ``operator_id`` on the resulting ``AssetOperation`` +
            ``AssetHistory`` row is the **leaver** (``person.id``).
            ``operator_id`` is NOT NULL on ``asset_operations``, and the
            leaver is the semantic trigger of every T+0 disposition —
            stamping the office admin / IT fallback would pollute their
            "operations I performed" audit queries with cron-driven
            writes they didn't initiate. ``detail.source`` keeps the
            cron-vs-human distinction.
          * ``recipient_id`` in the synthetic transfer item (consumed
            by ``_dispatch_post_advance_notifications``) is the
            **fallback_recipient** (office admin / IT fallback) for
            ``keep`` and degraded ``recycle_to_office`` — those branches
            don't move the asset to a new active person, but a real human
            still needs to see "this leaver's device is now held_by_leaver,
            it's on your plate". Routing this push to ``person.id`` would
            hit an SSO-suspended inbox.

        Self-borrow corner case (PR-#406 review fix):
          When the leaver is BOTH owner AND current borrower of an
          asset (rare but observed), we clear the borrow state up
          front so the downstream ``recycle_to_office`` service call
          doesn't trip its ``ASSET_BORROWED`` guard. The borrow side
          gets a separate audit history row so the timeline reads
          "borrow auto-returned at offboarding" + the disposition row,
          not a single confusing transfer.

        Three branches:

          * ``keep``   — leaver-keeps-device. Delegates to
            ``AssetService.mark_held_by_leaver``. Status flips to
            ``held_by_leaver``, ``assigned_to`` preserved.
          * ``recycle_to_office`` — default. ``office_id`` from the
            disposition entry, else resolved from the leaver's home
            office. No resolvable office DEGRADES to
            ``mark_held_by_leaver`` (auditable; IT recovers later).
          * ``transfer_to_person`` — explicit successor required.
            Invalid / missing successor falls back to
            ``fallback_recipient`` (manager → IT admin → fallback).
        """
        from uuid import UUID as _UUID  # noqa: PLC0415

        from app.domains.asset.models import (  # noqa: PLC0415
            AssetHistory,
            AssetOperation,
        )
        from app.domains.asset.service import (  # noqa: PLC0415
            AssetService,
            _generate_operation_number,
        )

        asset_svc = AssetService(self.db)
        action = disposition.get("action") or "recycle_to_office"
        reason_label = (
            f"T+0 disposition ({action}): leaver "
            f"{person.display_name or person.work_email}"
        )

        # Self-borrow guard. Before delegating to recycle_to_office /
        # mark_held_by_leaver (both reject borrowed=non-NULL), clear the
        # borrow side. Pure-borrower-only blockers (was_borrower AND
        # NOT was_owner) are handled in ``_handle_one_blocker`` and never
        # reach this method.
        if was_borrower and asset.borrowed_by == person.id:
            asset.borrowed_by = None
            asset.borrow_started_at = None
            asset.expected_return_date = None
            # The leaver is also the owner (otherwise the pure-borrower
            # branch upstream would have handled this). Restore the
            # personal-claim status so the downstream recycle / keep
            # service-call's source-state whitelist accepts the row.
            # Pre-clear leaves a "borrow auto-returned" history row;
            # the downstream call adds the recycle / keep history row.
            if asset.status == "borrowed":
                asset.status = "claimed"
            self.db.add(AssetHistory(
                asset_id=asset.id,
                action="borrow_return",
                from_user=person.id,
                to_user=prev_assigned_to,
                operator_id=person.id,
                notes=(
                    f"T+0 自动归还自借资产 (offboarding disposition pre-clear): "
                    f"leaver {person.display_name or person.work_email}"
                ),
            ))
            await self.db.flush()

        fallback_label = (
            fallback_recipient.display_name
            or fallback_recipient.work_email
            or "IT 处置 (T+0)"
        )

        if action == "keep":
            default_fallback = bool(disposition.get("_default_fallback"))
            source = (
                "offboarding_disposition_default"
                if default_fallback
                else "offboarding_disposition"
            )
            await asset_svc.mark_held_by_leaver(
                asset.id,
                reason=(
                    reason_label
                    + (" (默认兜底: 无人处置)" if default_fallback else "")
                ),
                operator_id=person.id,  # leaver is the semantic trigger
                source=source,
            )
            return True, {
                "type": "hardware",
                "label": (
                    blocker.label or asset.asset_name or str(asset.id)
                ),
                # Notification goes to the human who needs to know — NOT
                # the leaver's suspended inbox.
                "recipient_id": str(fallback_recipient.id),
                "recipient_label": fallback_label,
                "disposition_action": "keep",
                "default_fallback": default_fallback,
                "borrow_returned_to_owner": False,
                "original_lender_id": None,
                "asset_id": str(asset.id),
                "asset_tag": asset.asset_tag,
            }

        if action == "recycle_to_office":
            office_id_raw = disposition.get("office_id")
            office_id = None
            if office_id_raw:
                try:
                    office_id = _UUID(str(office_id_raw))
                except (TypeError, ValueError):
                    office_id = None
            if office_id is None:
                # Resolve from leaver's home office.
                home_office = await asset_svc.resolve_office_for_person(
                    person.id,
                )
                if home_office is not None:
                    office_id = home_office.id
            if office_id is None:
                # No usable office — degrade to held_by_leaver so the
                # asset doesn't silently flip into an unowned, unrouted
                # state. The IT admin clears it later.
                await asset_svc.mark_held_by_leaver(
                    asset.id,
                    reason=(
                        reason_label
                        + " (degraded: 无法解析归属办公室)"
                    ),
                    operator_id=person.id,
                    source="offboarding_disposition_degraded",
                )
                return True, {
                    "type": "hardware",
                    "label": (
                        blocker.label or asset.asset_name or str(asset.id)
                    ),
                    "recipient_id": str(fallback_recipient.id),
                    "recipient_label": fallback_label,
                    "disposition_action": "recycle_to_office_degraded",
                    "borrow_returned_to_owner": False,
                    "original_lender_id": None,
                }
            _, _op = await asset_svc.recycle_to_office(
                asset.id,
                office_id=office_id,
                reason=reason_label,
                operator_id=person.id,
                source="offboarding_disposition",
            )
            office = await self.db.get(AssetOfficeAddress, office_id)
            office_label = (
                office.office_name or office.city
                if office is not None
                else str(office_id)
            )
            # The device is being shipped to an office, so the
            # recipient is whoever administers the location that office
            # sits at — not the office's 当地 Admin, who is a contact
            # for the people there rather than for its hardware. Falls
            # back to the configured IT contact when that location has
            # no single accountable administrator.
            office_admin = (
                await resolve_office_location_primary_admin(self.db, office.id)
                if office is not None
                else None
            )
            return True, {
                "type": "hardware",
                "label": (
                    blocker.label or asset.asset_name or str(asset.id)
                ),
                "recipient_id": (
                    str(office_admin.id)
                    if office_admin is not None
                    else str(fallback_recipient.id)
                ),
                "recipient_label": f"🏢 {office_label}",
                "disposition_action": "recycle_to_office",
                "office_id": str(office_id),
                "borrow_returned_to_owner": False,
                "original_lender_id": None,
            }

        if action == "transfer_to_person":
            person_id_raw = disposition.get("person_id")
            recipient = fallback_recipient
            silent_retarget = False
            chosen_label = None
            if person_id_raw:
                try:
                    pid = _UUID(str(person_id_raw))
                    candidate = await self.db.get(IdentityPerson, pid)
                    if (
                        candidate is not None
                        and candidate.deleted_at is None
                        and candidate.employment_status == "active"
                    ):
                        recipient = candidate
                    elif candidate is not None:
                        # Admin picked someone who has since become
                        # inactive (offboarding race). Record the
                        # silent-retarget so the audit row makes the
                        # divergence visible to a reviewer instead of
                        # the asset just landing on the manager / IT
                        # admin with no trace of operator intent.
                        silent_retarget = True
                        chosen_label = (
                            candidate.work_email
                            or candidate.display_name
                            or str(candidate.id)
                        )
                except (TypeError, ValueError):
                    pass

            # Pre-launch P1 fix (Round 45, CR-416): when the chosen
            # successor is no longer active, write a structured warning
            # into the AssetOperation detail. Without this the admin
            # who picked "Carol" sees the asset land with "Dave"
            # (Carol's manager via fallback) and no obvious signal
            # that the system silently retargeted.

            # Incoming-hire reservation reconciliation (备机闭环 review
            # fix, 2026-06-12): a leaver's device may be earmarked for
            # an incoming hire (inherit_from_leaver). The disposition
            # is admin-configured truth, so it wins — but the earmark
            # must not survive as dual-state. fulfilled/superseded is
            # recorded in the op detail below.
            from app.domains.asset.reservation import (
                resolve_reservation_on_ownership_move,
            )

            reservation_outcome = resolve_reservation_on_ownership_move(
                asset, recipient.id,
            )
            asset.assigned_to = recipient.id
            asset.assigned_at = datetime.now(UTC)
            if was_borrower:
                # Pre-clear above already nulled borrowed_by for the
                # leaver-borrowed case; this re-applies for the rarer
                # "borrowed_by is a third party" subcase (the leaver
                # is owner, the device is currently on loan to someone
                # else). We do NOT move that loan to the recipient —
                # the new owner inherits the lend-out unchanged, mirror
                # of pre-PR behaviour.
                pass
            now = datetime.now(UTC)
            transfer_notes = (
                f"T+0 transfer_to_person disposition: leaver "
                f"{person.display_name or person.work_email} → "
                f"{recipient.display_name or recipient.work_email}"
            )
            self.db.add(AssetHistory(
                asset_id=asset.id,
                action="force_transfer",
                from_user=prev_assigned_to,
                to_user=asset.assigned_to,
                operator_id=person.id,
                notes=transfer_notes,
            ))
            # P1-6 fix: write the matching AssetOperation row so the
            # Operations admin page renders T+0 transfers alongside
            # manual recycle / release rows. Pre-PR-#406 review this
            # was history-only and the operation log under-counted T+0
            # actions.
            op_detail: dict[str, Any] = {
                "source": "offboarding_disposition",
                "disposition_action": "transfer_to_person",
                "leaver_person_id": str(person.id),
                "successor_person_id": str(recipient.id),
                "prior_assigned_to": (
                    str(prev_assigned_to) if prev_assigned_to else None
                ),
            }
            if reservation_outcome is not None:
                op_detail["hire_reservation"] = reservation_outcome
            if silent_retarget:
                op_detail["silent_retarget"] = True
                op_detail["originally_chosen_person_id"] = str(person_id_raw)
                op_detail["originally_chosen_label"] = chosen_label
                op_detail["retarget_reason"] = (
                    "chosen successor no longer active at T+0"
                )
            self.db.add(AssetOperation(
                operation_number=_generate_operation_number(),
                asset_id=asset.id,
                operation_type="transfer",
                status="confirmed",
                from_user_id=prev_assigned_to,
                to_user_id=recipient.id,
                operator_id=person.id,
                notes=transfer_notes[:2000],
                detail=op_detail,
                confirmed_at=now,
                confirmed_by=person.id,
            ))
            return True, {
                "type": "hardware",
                "label": (
                    blocker.label or asset.asset_name or str(asset.id)
                ),
                "recipient_id": str(recipient.id),
                "recipient_label": (
                    recipient.display_name or recipient.work_email
                ),
                "disposition_action": "transfer_to_person",
                "borrow_returned_to_owner": False,
                "original_lender_id": None,
            }

        # Unknown action — refuse the disposition, leave blocker
        # pending. This is a defensive branch; HardwareDispositionEntry
        # validator should have caught it.
        return False, None

    async def _mark_provisioning_task_done(
        self,
        blocker: OffboardingBlocker,
        *,
        now: datetime,
    ) -> None:
        """Best-effort mark of any linked provisioning task as completed.

        Prevents ``suspend_sso`` from re-attempting a connector call
        for a resource that is no longer the leaver's. Idempotent —
        the task may already be completed (e.g. Zoom auto_revoke
        succeeded earlier in the tick) in which case we leave it
        alone.

        Pre-launch P2 fix (Round 1, CR-007): take ``SELECT ... FOR
        UPDATE`` and skip when the task is in ANY non-pre-flight
        state. Both the provisioning dispatcher cron and the advance
        cron run on a 15-min cadence; collisions are normal. The
        prior code only refused ``status==COMPLETED``, so an
        ``IN_PROGRESS`` task would be clobbered to COMPLETED while
        the original worker was still mid-vendor-call, with no
        ``external_ref`` and an inconsistent state versus the vendor.
        """
        if blocker.provisioning_task_id is None:
            return
        from app.domains.provisioning.models import (  # noqa: PLC0415
            STATUS_CANCELLED,
            STATUS_COMPLETED,
            STATUS_FAILED,
            STATUS_IN_PROGRESS,
            ProvisioningTask,
        )
        task = (
            await self.db.execute(
                select(ProvisioningTask)
                .where(ProvisioningTask.id == blocker.provisioning_task_id)
                .with_for_update()
            )
        ).scalar_one_or_none()
        if task is None:
            return
        if task.status in (
            STATUS_IN_PROGRESS,
            STATUS_COMPLETED,
            STATUS_FAILED,
            STATUS_CANCELLED,
        ):
            return
        task.status = STATUS_COMPLETED
        task.completed_at = now
        task.error_message = None
        blocker.resolution = RESO_REVOKED_AUTO

        # Pre-launch P2 fix: when a software-seat provisioning task is
        # force-completed here, the SoftwareSeatAllocation row is never
        # updated by the normal revoke pipeline (the connector was
        # bypassed). Flip the allocation to "revoked" now so that:
        #   1. Billing crons stop counting the seat as active.
        #   2. post_leaver_verify doesn't raise a false-positive incident.
        #   3. The admin UI shows the correct "revoked" state.
        # task.target_type == "seat" identifies a seat revoke task whose
        # target_id is the SoftwareSeatAllocation primary key.
        if task.action == "revoke" and task.target_type == "seat" and task.target_id:
            from app.domains.asset.models import SoftwareSeatAllocation  # noqa: PLC0415
            alloc = await self.db.get(SoftwareSeatAllocation, task.target_id)
            if alloc is not None and alloc.status != "revoked":
                alloc.status = "revoked"
                alloc.revoked_at = now

    async def _handle_one_blocker(
        self,
        *,
        blocker: OffboardingBlocker,
        person: IdentityPerson,
        process: OffboardingProcess,
        default_new_owner: IdentityPerson,
        now: datetime,
        sso_revoked_collector: list[dict[str, Any]] | None = None,
    ) -> tuple[bool, dict[str, Any] | None]:
        """Resolve one blocker. Caller wraps in begin_nested savepoint.

        Returns ``(moved, transferred_item)``:
          * ``moved=False, item=None`` — underlying resource missing /
            disposed / no longer owned by leaver. Caller skips
            synthetic transfer, blocker stays pending.
          * ``moved=True, item=None`` — direct-revoke branch already
            set ``blocker.resolution`` and ``blocker.notes``; no
            synthetic transfer needed (e.g. SSO grant, cloud access
            grant, VPN application, AI tool application — these are
            people-bound and revoked at source).
          * ``moved=True, item={...}`` — branch mutated state and
            wants a synthetic ``AssetTransferRequest`` audit row.
            ``item['recipient_id']`` is the chosen new owner (office
            admin for hardware, default fallback for everything else).
        """
        resource_kind = blocker.resource_type
        resource_id = blocker.resource_id
        if resource_id is None:
            return False, None

        # Recipient resolution — hardware tries office admin first
        # (and explicitly bypasses the IT-prefilled handover person:
        # the asset-fallback-matrix promises laptops only flow to
        # Office Admin / leaver-retain). Everything else honours the
        # IT-set ``OffboardingProcess.default_handover_person_id``
        # before climbing the manager chain — this is the "有交接人
        # 给交接人,无交接给 lead" rule the UI matrix exposes.
        chosen_owner: IdentityPerson = default_new_owner
        if resource_kind == RES_HARDWARE:
            asset_for_recipient = await self.db.get(
                AssetItem, resource_id,
            )
            chosen_owner = await self._resolve_hardware_recipient(
                asset=asset_for_recipient, default=default_new_owner,
            )
        elif process.default_handover_person_id is not None:
            # Non-hardware path: short-circuit to the IT-set handover
            # person when they're still receivable. Identical guards
            # to ``_resolve_fallback_person``'s preferred-person check
            # — active, not soft-deleted, and not the leaver themself.
            preferred_id = process.default_handover_person_id
            if preferred_id != person.id:
                preferred = await self.db.get(IdentityPerson, preferred_id)
                if (
                    preferred is not None
                    and preferred.deleted_at is None
                    and preferred.employment_status == "active"
                ):
                    chosen_owner = preferred

        recipient_label = (
            chosen_owner.display_name or chosen_owner.work_email
        )

        if resource_kind == RES_HARDWARE:
            asset = await self.db.get(AssetItem, resource_id)
            if asset is None:
                return False, None
            # Distinguish two ownership patterns:
            #
            # 1. Leaver is the OWNER (``assigned_to == leaver``).
            #    Disposition-driven decision (migration 184): the admin
            #    may have picked ``keep`` / ``recycle_to_office`` /
            #    ``transfer_to_person`` per asset; we honour that here.
            #    Default (no disposition entry) = ``recycle_to_office``
            #    to the leaver's home office, so the asset stops piling
            #    on the office admin / manager and surfaces correctly in
            #    that office's next inventory cycle.
            #
            # 2. Leaver is the BORROWER of someone else's asset
            #    (``borrowed_by == leaver`` while ``assigned_to``
            #    points elsewhere). Force-RETURN the borrow to the
            #    original lender — disposition does not apply because
            #    the asset isn't the leaver's to dispose of.
            was_owner = asset.assigned_to == person.id
            was_borrower = asset.borrowed_by == person.id
            if not (was_owner or was_borrower):
                return False, None
            original_lender_id = (
                asset.assigned_to
                if was_borrower and not was_owner else None
            )
            prev_assigned_to = asset.assigned_to

            # Pure-borrower path is unchanged: return to lender.
            if was_borrower and not was_owner:
                asset.borrowed_by = None
                asset.borrow_started_at = None
                asset.expected_return_date = None
                if asset.status == "borrowed":
                    # Return the asset to the lender's canonical state.
                    # ``"in_use"`` is NOT a member of ``ASSET_IN_USE_STATUSES``
                    # — using it here silently corrupted the status enum and
                    # caused the lender's own future offboarding scan to miss
                    # this asset (post_leaver_service line 155 was coupled to
                    # the same typo). Canonical state for an asset that has
                    # a lender = ``"assigned"`` (or ``"idle"`` when the
                    # original lent-asset had no owner).
                    asset.status = (
                        "assigned" if asset.assigned_to is not None else "idle"
                    )
                from app.domains.asset.models import (  # noqa: PLC0415
                    AssetHistory,
                )
                self.db.add(AssetHistory(
                    asset_id=asset.id,
                    action="borrow_return",
                    from_user=person.id,
                    to_user=prev_assigned_to,
                    operator_id=chosen_owner.id,
                    notes=(
                        f"T+0 自动归还借出资产: leaver "
                        f"{person.display_name or person.work_email} → 原主"
                    ),
                ))
                return True, {
                    "type": "hardware",
                    "label": (
                        blocker.label or asset.asset_name or str(asset.id)
                    ),
                    "recipient_id": (
                        str(original_lender_id) if original_lender_id else
                        str(chosen_owner.id)
                    ),
                    "recipient_label": "原借出方",
                    "borrow_returned_to_owner": True,
                    "original_lender_id": (
                        str(original_lender_id)
                        if original_lender_id else None
                    ),
                }

            # Owner path — disposition-aware (migration 184).
            disposition = self._resolve_hardware_disposition(
                process=process,
                asset=asset,
                fallback_recipient=chosen_owner,
            )
            return await self._apply_hardware_disposition(
                blocker=blocker,
                asset=asset,
                person=person,
                disposition=disposition,
                fallback_recipient=chosen_owner,
                prev_assigned_to=prev_assigned_to,
                was_borrower=was_borrower,
            )

        if resource_kind == RES_API_KEY:
            assignment = await self.db.get(ApiKeyAssignment, resource_id)
            if assignment is None or assignment.assigned_to != person.id:
                return False, None
            # Round-49: delegate to ``ApiKeyService.apply_transfer`` so
            # the force-transfer carries the full side-effect chain
            # (cost-allocation rule swap, AWS IAM tag rewrite, audit
            # row) that the voluntary accept path uses. The prior code
            # just flipped ``assigned_to`` with zero audit, leaving
            # cloud IAM tagged with the leaver's email, the
            # cost-allocation rule still pointing at the leaver, and
            # no forensic trail. ``apply_transfer`` accepts
            # ``transfer_request_id=None`` for this force-transfer
            # case — trace is derived from the offboarding process id.
            from app.domains.apikey.service import ApiKeyService  # noqa: PLC0415
            # SAVEPOINT, not a bare ``except``. This one matters more than
            # most: the FALLBACK below writes (``assigned_to`` + an audit
            # row) on the same session. If ``apply_transfer`` died on a
            # flush the transaction is already dead, so the fallback that
            # exists to stop the offboarding getting stuck would itself
            # raise — the offboarding gets stuck AND the audit trail this
            # comment promises is never written. The savepoint rolls back
            # only the failed transfer and hands the session back usable,
            # which is the precondition the fallback silently assumed.
            async with best_effort(
                self.db,
                label=f"api-key force-transfer for assignment {assignment.id}",
                log=logger,
            ) as transfer_step:
                await ApiKeyService(self.db).apply_transfer(
                    assignment_id=assignment.id,
                    new_owner_id=chosen_owner.id,
                    transfer_request_id=None,
                    offboarding_process_id=process.id,
                )
            if not transfer_step.ok:
                # apply_transfer's contract is "best-effort cloud tag
                # rewrite, raise on local-row mis-state". A raise here
                # means the cost-allocation rule swap or row flip
                # failed — fall back to the bare ``assigned_to``
                # write so the offboarding doesn't get permanently
                # stuck, but write an inline audit row so reconciliation
                # is traceable.
                logger.warning(
                    "advance_service: API-key force-transfer "
                    "delegate to ApiKeyService.apply_transfer raised "
                    "for assignment=%s; falling back to direct "
                    "assigned_to write (no cost-rule swap / cloud "
                    "tag) — manual reconciliation required: %s",
                    assignment.id, transfer_step.reason(),
                )
                assignment.assigned_to = chosen_owner.id
                from app.audit.service import AuditService as _Audit  # noqa: PLC0415
                await _Audit(self.db).log(
                    trace_id=str(process.id),
                    actor_type="system",
                    actor_user_id=None,
                    action="apikey.transfer.force_fallback",
                    status="partial",
                    resource_type="api_key_assignment",
                    resource_id=str(assignment.id),
                    detail={
                        "previous_owner_person_id": str(person.id),
                        "new_owner_person_id": str(chosen_owner.id),
                        "leaver_person_id": str(person.id),
                        "offboarding_process_id": str(process.id),
                        "fallback_reason": (
                            "ApiKeyService.apply_transfer raised; "
                            "fell back to bare assigned_to write — "
                            "cost-rule + cloud-tag NOT swapped"
                        ),
                    },
                )
            return True, {
                "type": "api_key",
                "label": blocker.label or str(assignment.id),
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        # Network-asset force-transfer (domain / DNS / cloud_server /
        # load_balancer / cdn_domain). Each branch (a) verifies the
        # row still points at the leaver (skip stale blocker rows),
        # (b) flips ``owner_person_id`` to the chosen successor,
        # (c) writes a ``cmdb_network_owner_history`` row IN the
        # same tx so the per-resource ownership trail is preserved,
        # (d) writes an ``audit_logs`` row attributing the change to
        # the offboarding force-transfer (the resource row's
        # ``last_modified_*`` is overwritten on every sync upsert so
        # it's not a forensic source).
        from app.audit.service import AuditService  # noqa: PLC0415
        from app.domains.cmdb.owner_history import (  # noqa: PLC0415
            CHANGE_REASON_OFFBOARDING_FORCE_TRANSFER,
            RESOURCE_TYPE_CDN_DOMAIN,
            RESOURCE_TYPE_CERTIFICATE,
            RESOURCE_TYPE_DNS_RECORD,
            RESOURCE_TYPE_DOMAIN,
            RESOURCE_TYPE_LOAD_BALANCER,
            record_owner_change,
        )

        async def _record_owner_force_transfer_audit(
            *,
            resource_type: str,
            resource_id_str: str,
            previous_owner: str,
            new_owner: str,
            audit_action: str,
            label: str,
        ) -> None:
            """Audit-only sibling of ``_record_network_transfer`` for
            resource types without a domain-side owner-history table
            (cloud_scope / cloud_iam_user / software_owner /
            software_it_contact / domain_registration /
            google_orphan_owner). Without this row, governance queries
            on "who changed this owner" return nothing — the only
            trace was the synthetic ``AssetTransferRequest`` in the
            offboarding domain, invisible from the CMDB/Google/SSO
            governance surface.
            """
            audit_svc = AuditService(self.db)
            await audit_svc.log(
                trace_id=str(process.id),
                actor_type="system",
                actor_user_id=None,
                action=audit_action,
                status="success",
                resource_type=resource_type,
                resource_id=resource_id_str,
                detail={
                    "previous_owner_person_id": previous_owner,
                    "new_owner_person_id": new_owner,
                    "leaver_person_id": previous_owner,
                    "offboarding_process_id": str(process.id),
                    "blocker_id": str(blocker.id),
                    "label": label,
                    "via": "offboarding_force_transfer",
                },
            )

        async def _record_network_transfer(
            *,
            resource_type: str,
            resource_id_str: str,
            previous_owner: str,
            new_owner: str,
            audit_action: str,
            label: str,
        ) -> None:
            await record_owner_change(
                self.db,
                resource_type=resource_type,
                resource_id=resource_id_str,
                previous_owner_person_id=previous_owner,
                new_owner_person_id=new_owner,
                actor_person_id=None,  # system-initiated
                change_reason=CHANGE_REASON_OFFBOARDING_FORCE_TRANSFER,
                trace_id=str(process.id),
                context={
                    "offboarding_process_id": str(process.id),
                    "leaver_person_id": previous_owner,
                    "blocker_id": str(blocker.id),
                },
            )
            audit_svc = AuditService(self.db)
            await audit_svc.log(
                trace_id=str(process.id),
                actor_type="system",
                actor_user_id=None,
                action=audit_action,
                status="success",
                resource_type=resource_type,
                resource_id=resource_id_str,
                detail={
                    "previous_owner_person_id": previous_owner,
                    "new_owner_person_id": new_owner,
                    "leaver_person_id": previous_owner,
                    "offboarding_process_id": str(process.id),
                    "blocker_id": str(blocker.id),
                    "label": label,
                    "via": "offboarding_force_transfer",
                },
            )

        if resource_kind == RES_DOMAIN:
            dom = await self.db.get(CmdbDomain, resource_id)
            if dom is None or dom.owner_person_id != str(person.id):
                return False, None
            previous_owner = dom.owner_person_id
            dom.owner_person_id = str(chosen_owner.id)
            label = blocker.label or dom.domain_name or str(dom.id)
            await _record_network_transfer(
                resource_type=RESOURCE_TYPE_DOMAIN,
                resource_id_str=str(dom.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.domain.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "domain",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_DNS_RECORD:
            rec = await self.db.get(CmdbDnsRecord, resource_id)
            if rec is None or rec.owner_person_id != str(person.id):
                return False, None
            previous_owner = rec.owner_person_id
            rec.owner_person_id = str(chosen_owner.id)
            label = blocker.label or str(rec.id)
            await _record_network_transfer(
                resource_type=RESOURCE_TYPE_DNS_RECORD,
                resource_id_str=str(rec.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.dns_record.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "dns_record",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_CLOUD_SERVER:
            srv = await self.db.get(CmdbServer, resource_id)
            if srv is None or srv.owner_person_id != str(person.id):
                return False, None
            previous_owner = srv.owner_person_id
            srv.owner_person_id = str(chosen_owner.id)
            label = blocker.label or srv.name or str(srv.id)
            # Pre-launch P1 fix (Round 45, CR-205): write a per-server
            # owner history row alongside the audit log. cloud_server
            # is the highest-cost cmdb resource (instance owner = cost
            # owner). Without per-resource chronology, the governance
            # query "show me every owner change on cmdb_servers.<id>"
            # returned a generic audit_logs row only — no easy join with
            # other history surfaces, no rendering in the standard
            # owner-history admin view.
            from app.domains.cmdb.owner_history import (  # noqa: PLC0415
                CHANGE_REASON_OFFBOARDING_FORCE_TRANSFER,
                RESOURCE_TYPE_CLOUD_SERVER,
                record_owner_change,
            )

            await record_owner_change(
                self.db,
                resource_type=RESOURCE_TYPE_CLOUD_SERVER,
                resource_id=str(srv.id),
                previous_owner_person_id=previous_owner,
                new_owner_person_id=str(chosen_owner.id),
                actor_person_id=None,
                change_reason=CHANGE_REASON_OFFBOARDING_FORCE_TRANSFER,
                trace_id=str(process.id),
                context={
                    "offboarding_process_id": str(process.id),
                    "blocker_id": str(blocker.id),
                    "label": label,
                    "via": "offboarding_force_transfer",
                },
            )
            audit_svc = AuditService(self.db)
            await audit_svc.log(
                trace_id=str(process.id),
                actor_type="system",
                actor_user_id=None,
                action="cmdb.cloud_server.owner_force_transferred",
                status="success",
                resource_type="cloud_server",
                resource_id=str(srv.id),
                detail={
                    "previous_owner_person_id": previous_owner,
                    "new_owner_person_id": str(chosen_owner.id),
                    "leaver_person_id": previous_owner,
                    "offboarding_process_id": str(process.id),
                    "blocker_id": str(blocker.id),
                    "label": label,
                    "via": "offboarding_force_transfer",
                },
            )
            return True, {
                "type": "cloud_server",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_LOAD_BALANCER:
            lb = await self.db.get(CmdbLoadBalancer, resource_id)
            if lb is None or lb.owner_person_id != str(person.id):
                return False, None
            previous_owner = lb.owner_person_id
            lb.owner_person_id = str(chosen_owner.id)
            # Keep ``owner_raw`` in sync — it's a UI display fallback
            # for tables that show the legacy raw text.
            new_owner_label = (
                chosen_owner.work_email or chosen_owner.display_name
            )
            if new_owner_label:
                lb.owner_raw = new_owner_label
            label = blocker.label or lb.name or str(lb.id)
            await _record_network_transfer(
                resource_type=RESOURCE_TYPE_LOAD_BALANCER,
                resource_id_str=str(lb.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.lb.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "load_balancer",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_CDN_DOMAIN:
            cdn = await self.db.get(CmdbCdnDomain, resource_id)
            if cdn is None or cdn.owner_person_id != str(person.id):
                return False, None
            previous_owner = cdn.owner_person_id
            cdn.owner_person_id = str(chosen_owner.id)
            new_owner_label = (
                chosen_owner.work_email or chosen_owner.display_name
            )
            if new_owner_label:
                cdn.owner_raw = new_owner_label
            label = blocker.label or cdn.domain_name or str(cdn.id)
            await _record_network_transfer(
                resource_type=RESOURCE_TYPE_CDN_DOMAIN,
                resource_id_str=str(cdn.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.cdn_domain.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "cdn_domain",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_EMAIL_GROUP:
            from app.domains.email_group.models import (  # noqa: PLC0415
                EmailGroup,
            )
            from app.domains.email_group.service import (  # noqa: PLC0415
                EmailGroupService,
            )
            eg = await self.db.get(EmailGroup, resource_id)
            if eg is None:
                return False, None
            if eg.status == "active":
                eg_service = EmailGroupService(self.db)
                await eg_service.transfer_group_owner(
                    group=eg,
                    from_person=person,
                    to_person=chosen_owner,
                    actor_user_id=chosen_owner.id,
                    trace_id=str(process.id),
                )
                return True, {
                    "type": "email_group",
                    "label": blocker.label or eg.email or str(eg.id),
                    "recipient_id": str(chosen_owner.id),
                    "recipient_label": recipient_label,
                }
            # Inactive group — treat as resolved without a real transfer.
            # Pre-launch P2 fix: must set blocker.resolution so that
            # _recompute_status doesn't count this blocker as pending and
            # permanently block the leaver transition.
            from app.domains.offboarding.models import RESO_SKIPPED  # noqa: PLC0415
            blocker.resolution = RESO_SKIPPED
            blocker.notes = (
                (blocker.notes or "")
                + "\n[离职] email group 已停用, 自动跳过"
            ).strip()
            return True, {
                "type": "email_group",
                "label": blocker.label or str(resource_id),
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_SOFTWARE_OWNER:
            sw = await self.db.get(AssetSoftwareCatalog, resource_id)
            if sw is None or sw.owner_person_id != person.id:
                return False, None
            previous_owner = str(sw.owner_person_id)
            sw.owner_person_id = chosen_owner.id
            label = blocker.label or sw.name or str(sw.id)
            await _record_owner_force_transfer_audit(
                resource_type="asset_software_catalog",
                resource_id_str=str(sw.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="software.owner.force_transferred",
                label=label,
            )
            return True, {
                "type": "software_owner",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_SOFTWARE_IT_CONTACT:
            sw = await self.db.get(AssetSoftwareCatalog, resource_id)
            if sw is None or sw.it_contact_person_id != person.id:
                return False, None
            previous_owner = str(sw.it_contact_person_id)
            sw.it_contact_person_id = chosen_owner.id
            label = blocker.label or sw.name or str(sw.id)
            await _record_owner_force_transfer_audit(
                resource_type="asset_software_catalog",
                resource_id_str=str(sw.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="software.it_contact.force_transferred",
                label=label,
            )
            return True, {
                "type": "software_it_contact",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_CLOUD_SCOPE_OWNER:
            scope = await self.db.get(CmdbCloudScope, resource_id)
            if scope is None or scope.owner_person_id != str(person.id):
                return False, None
            previous_owner = str(scope.owner_person_id)
            scope.owner_person_id = str(chosen_owner.id)
            # The handover IS the confirmation (2026-08-24 product
            # decision). The successor already gets the
            # "[自动转让] … N 项资产已转给你" push naming this
            # subscription, so starting their 180-day clock here — rather
            # than leaving it NULL, which reads as "never confirmed" and
            # makes them immediately due — means they are told once, not
            # told and then asked again days later about the same thing.
            # Mirrors ``assign_scope_owner``, which has always bumped this
            # on a normal transfer; only the offboarding path was missing
            # it because it rewrites ``owner_person_id`` directly.
            scope.last_owner_confirmed_at = now
            label = (
                blocker.label or scope.display_name or str(scope.id)
            )
            await _record_owner_force_transfer_audit(
                resource_type="cmdb_cloud_scope",
                resource_id_str=str(scope.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.cloud_scope.owner_force_transferred",
                label=label,
            )
            # Release any owner-confirmation prompt still outstanding
            # against the LEAVER. This path rewrites ``owner_person_id``
            # directly instead of going through ``assign_scope_owner``,
            # so nothing else closes that row — and an open pending row
            # (a) makes ``list_scopes_due_semiannual`` skip the scope
            # forever, so the new owner is never asked to confirm, and
            # (b) keeps the daily reminder sweep pushing a card addressed
            # to someone who has left, every morning. Same session, same
            # transaction as the ownership rewrite: either both land or
            # neither does.
            from app.domains.cmdb.owner_confirmation_service import (  # noqa: PLC0415
                OwnerConfirmationService,
            )
            await OwnerConfirmationService(
                self.db,
            ).void_pending_for_owner_change(
                str(scope.id),
                new_owner_person_id=str(chosen_owner.id),
                trace_id=str(process.id),
                via="offboarding_force_transfer",
            )
            return True, {
                "type": "cloud_scope_owner",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_IDENTITY_CONSUMER_OWNER:
            # Defensive: the unconditional T+0 sweep in ``run_tick`` (see
            # ``_disable_unclaimed_consumers``) normally resolves these
            # before the bucket split, so a consumer blocker should not
            # reach the owner-transfer machinery. Kept as a single-source
            # fallback in case one is ever routed here.
            disabled = await self._disable_unclaimed_consumer(
                blocker, person, process,
            )
            return disabled, None

        if resource_kind == RES_CLOUD_IAM_USER:
            # Move the Helix-side ownership FK on cmdb_iam_users to the
            # fallback person. The actual IAM credential on the cloud
            # side keeps existing — only the Helix-side responsible-human
            # pointer moves. ``revoked_at IS NULL`` already filtered by
            # the scanner, but re-check here so a row that was revoked
            # between scan and T+0+24h is treated as already-resolved
            # (return False so the blocker stays pending for the next
            # tick to garbage-collect via _resync_blockers rather than
            # mark a synthetic transfer for a dead resource).
            iam_user = await self.db.get(CmdbIamUser, resource_id)
            if (
                iam_user is None
                or iam_user.deleted_at is not None
                or iam_user.revoked_at is not None
                or iam_user.owner_person_id != str(person.id)
            ):
                return False, None
            previous_owner = str(iam_user.owner_person_id)
            iam_user.owner_person_id = str(chosen_owner.id)
            label = (
                blocker.label
                or iam_user.user_name
                or iam_user.arn
                or str(iam_user.id)
            )
            await _record_owner_force_transfer_audit(
                resource_type="cmdb_iam_user",
                resource_id_str=str(iam_user.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.iam_user.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "cloud_iam_user",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_DOMAIN_REGISTRATION:
            from app.domains.domain_registration.models import (  # noqa: PLC0415
                DomainRegistration,
            )
            reg = await self.db.get(DomainRegistration, resource_id)
            if reg is None or reg.person_id != str(person.id):
                return False, None
            previous_owner = str(reg.person_id)
            reg.person_id = str(chosen_owner.id)
            label = blocker.label or reg.domain_name or str(reg.id)
            await _record_owner_force_transfer_audit(
                resource_type="domain_registration",
                resource_id_str=str(reg.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="domain_registration.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "domain_registration",
                # ``domain_name`` is a non-nullable column, so direct
                # access is safe — the prior ``getattr(reg, "domain_name", None)``
                # would silently swallow a future column rename, which is
                # exactly the bug class that bit PRs #122/127/128/129
                # (no-getattr-fallback rule, MEMORY.md). Round-17 audit.
                "label": blocker.label or reg.domain_name or str(reg.id),
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_SYSTEM_OWNER:
            # Registry ownership. The registry's own backup owner beats
            # the generic manager fallback: they were named in advance
            # precisely for this, and handing a live system to a manager
            # who has never touched it is how a system becomes ownerless
            # again at the next reorg.
            from app.core.coordinators import (  # noqa: PLC0415
                system_registry_coordinator as _registry,
            )

            owned_rows = await _registry.systems_owned_by(
                self.db, person.id,
            )
            row = next(
                (r for r in owned_rows if r["id"] == str(resource_id)), None,
            )
            if row is None:
                # No longer owned by the leaver (transferred manually
                # between scan and T+0+24h) — leave the blocker pending
                # for ``_resync_blockers`` to garbage-collect.
                return False, None
            system_recipient = chosen_owner
            backup_id = row["backup_owner_person_id"]
            if row["is_primary_owner"] and backup_id:
                backup = await self.db.get(IdentityPerson, UUID(backup_id))
                if (
                    backup is not None
                    and backup.deleted_at is None
                    and backup.employment_status == "active"
                    and backup.id != person.id
                ):
                    system_recipient = backup
            moved = await _registry.handover_systems(
                self.db,
                from_person_id=person.id,
                to_person_id=system_recipient.id,
                actor_person_id=None,
                note="离职强制交接 (T+0+24h)",
                system_ids=[resource_id],
            )
            if not moved:
                return False, None
            label = blocker.label or row["name"]
            if not row["is_primary_owner"]:
                # Backup-slot-only row: nothing was transferred, the
                # leaver was REMOVED as the emergency contact. Recording
                # it as an ownership transfer would put a false pair of
                # owners in the audit trail, and returning a synthetic
                # transfer item would invent a recipient who received
                # nothing.
                # Written directly rather than through
                # ``_record_owner_force_transfer_audit``: that helper's
                # detail shape asserts a new owner, and there isn't one.
                await AuditService(self.db).log(
                    trace_id=str(process.id),
                    actor_type="system",
                    actor_user_id=None,
                    action="system_registry.backup_owner_cleared",
                    status="success",
                    resource_type="registry_system",
                    resource_id=str(resource_id),
                    detail={
                        "cleared_person_id": str(person.id),
                        "leaver_person_id": str(person.id),
                        "offboarding_process_id": str(process.id),
                        "blocker_id": str(blocker.id),
                        "label": label,
                        "via": "offboarding_force_transfer",
                    },
                )
                return True, None
            await _record_owner_force_transfer_audit(
                resource_type="registry_system",
                resource_id_str=str(resource_id),
                previous_owner=str(person.id),
                new_owner=str(system_recipient.id),
                audit_action="system_registry.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "system_owner",
                "label": label,
                "recipient_id": str(system_recipient.id),
                "recipient_label": (
                    system_recipient.display_name
                    or system_recipient.work_email
                ),
            }

        if resource_kind == RES_GOOGLE_ORPHAN_OWNER:
            from app.domains.google_workspace.models import (  # noqa: PLC0415
                GoogleOrphanAccount,
            )
            orphan = await self.db.get(GoogleOrphanAccount, resource_id)
            # Column is ``owner_user_id`` (Uuid, not str). The previous
            # ``getattr(orphan, "owner_person_id", ...)`` always returned
            # None and the assignment wrote a transient Python attribute
            # that never persisted — orphan force-transfer was a silent
            # no-op for every leaver who owned a Google shared mailbox.
            if orphan is None or orphan.owner_user_id != person.id:
                return False, None
            previous_owner = str(orphan.owner_user_id)
            orphan.owner_user_id = chosen_owner.id
            label = blocker.label or orphan.primary_email or str(orphan.id)
            await _record_owner_force_transfer_audit(
                resource_type="google_orphan_account",
                resource_id_str=str(orphan.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="google.orphan_account.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "google_orphan_owner",
                # ``primary_email`` is a non-nullable column, so direct
                # access is safe (no-getattr-fallback rule).
                "label": blocker.label or orphan.primary_email or str(orphan.id),
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        # ── Direct-revoke branches: caller signals via ``item is None``
        #    that no synthetic transfer is needed. ``moved=True`` so
        #    the caller marks the blocker counted but skips the
        #    AssetTransferRequest INSERT. ──

        if resource_kind == RES_SSO_GRANT:
            from app.domains.sso.models import SsoGrant  # noqa: PLC0415
            from app.domains.sso.service import SsoGrantService  # noqa: PLC0415
            # Round-58 (2026-05-30) — security scan C-1: previously the
            # T+0 path mutated ``grant.status = REVOKED`` directly,
            # bypassing ``SsoService.revoke``. That skipped TWO safety
            # rails:
            #   (1) the ``sso.grant.revoke`` audit row,
            #   (2) the system-owner + IT-admin "[Action Required]
            #       Disable SSO access for {person}" push — without
            #       which Helix marks the grant ``revoked`` while the
            #       real SAML/SSO console (Google admin / Atlassian /
            #       Okta) still permits live login for hours-to-days
            #       until somebody happens to act. SSO revoke in Helix
            #       is DB-only; the cloud-side console action is the
            #       actual cut-off and the notification is what makes
            #       sure it happens.
            # Pre-Round-58 the direct-mutation lived inside the 24h grace
            # hold so the bug bit only on day 2+ when grace expired.
            # Round-58 moves SSO into the IMMEDIATE_REVOKE bucket — so
            # the wrong-path now fires on every leaver T+0. Replace with
            # the proper service call: same revoke semantics + audit +
            # notification + FOR UPDATE lock (M-1).
            # H-1 (security scan): cast resource_id to str — SsoGrant.id
            # is ``Mapped[str]`` on a String(36) column. Pass uuid.UUID
            # against a String PK on TiDB+asyncmy returns 0 rows
            # silently (memory ref ``reference_asyncmy_uuid_bind``).
            # VPN / AI branches already cast; SSO didn't.
            grant_id_str = str(resource_id)
            grant = await self.db.get(SsoGrant, grant_id_str)
            if grant is None or grant.person_id != str(person.id):
                return False, None
            already_revoked = False
            try:
                # Round-59 follow-up (2026-05-31): notify_admins=False
                # — SsoGrantService.revoke fans out one card per IT
                # admin per grant when called per-grant. At T+0 with N
                # SSO grants and M IT admins that's N×M cards per
                # leaver. The bulk caller (_force_transfer_all)
                # collects revoked grants via sso_revoked_collector and
                # emits ONE coalesced summary card per admin per leaver
                # after the loop. The per-system owner card inside
                # revoke() still fires (system_owner_person_id varies
                # per grant — natural fan-out bound).
                from app.domains.provisioning.models import (  # noqa: PLC0415
                    SOURCE_OFFBOARDING as _PROV_SOURCE_OFFBOARDING,
                )
                # ``task_source`` stamps the removal task (migration 603) as
                # offboarding work so the queue and the leaver dashboards
                # file it with the departure, not as an ad-hoc admin revoke.
                await SsoGrantService(self.db).revoke(
                    grant_id=grant_id_str,
                    actor=default_new_owner,
                    reason="offboarding",
                    trace_id=str(process.id),
                    notify_admins=False,
                    task_source=_PROV_SOURCE_OFFBOARDING,
                )
            except AppException as exc:
                # SsoService.revoke raises SSO_GRANT_NOT_REVOKABLE when
                # the grant is not in ``approved`` state — typically a
                # leaver whose grant was already revoked via another
                # path (Feishu cancel webhook / handover). Treat as
                # no-op: blocker resolves, no compliance incident, no
                # duplicate audit. Do NOT add to coalesced summary —
                # nothing was actually revoked this tick.
                if exc.code == "SSO_GRANT_NOT_REVOKABLE":
                    already_revoked = True
                else:
                    raise
            if not already_revoked and sso_revoked_collector is not None:
                sso_revoked_collector.append({
                    "grant_id": grant_id_str,
                    "system_name": grant.system_name or "(unnamed system)",
                })
            blocker.resolution = RESO_REVOKED_AUTO
            blocker.resolution_ref = (
                f"offboarding_revoke:sso_grant:{grant_id_str}"
            )
            blocker.notes = (
                (blocker.notes or "")
                + "\n[离职] 自动 revoke SSO 授权"
            ).strip()
            return True, None

        if resource_kind == RES_VPN_APPLICATION:
            from app.domains.vpn_application.models import (  # noqa: PLC0415
                VPN_APP_STATUS_APPROVED,
                VpnApplication,
            )
            from app.domains.vpn_application.service import (  # noqa: PLC0415
                VpnApplicationService,
            )
            vpn_row = await self.db.get(VpnApplication, str(resource_id))
            if (
                vpn_row is None
                or vpn_row.person_id != str(person.id)
                or vpn_row.status != VPN_APP_STATUS_APPROVED
            ):
                return False, None
            # NOTE: exceptions intentionally propagate to the outer
            # _force_transfer_all SAVEPOINT handler, which records the
            # failure on the blocker's last_error_* columns and
            # leaves it in ``pending`` so the next tick retries and the
            # triage_bucket logic promotes it to ``needs_it``. Pre-148
            # behaviour swallowed driver errors here and still marked
            # the blocker ``revoked_auto`` — which silently obscured
            # vendor outages and left the leaver flagged "done" with a
            # still-active VPN session.
            await VpnApplicationService(self.db).revoke(
                app_id=str(vpn_row.id),
                actor=default_new_owner,
                reason="offboarding",
                trace_id=str(process.id),
            )
            blocker.resolution = RESO_REVOKED_AUTO
            blocker.resolution_ref = (
                f"offboarding_revoke:vpn_application:{vpn_row.id}"
            )
            blocker.notes = (
                (blocker.notes or "") + "\n[离职] 自动 revoke VPN 申请"
            ).strip()
            return True, None

        if resource_kind == RES_GITHUB_ORG_MEMBERSHIP:
            from app.domains.github_org import (  # noqa: PLC0415
                constants as gh_const,
            )
            from app.domains.github_org import execution as gh_exec
            from app.domains.github_org.models import (  # noqa: PLC0415
                GithubOrg,
                GithubOrgMembership,
            )

            membership = await self.db.get(
                GithubOrgMembership, str(resource_id),
            )
            if (
                membership is None
                or membership.person_id != str(person.id)
                or membership.state not in gh_const.MEMBERSHIP_LIVE_STATES
            ):
                # 已经不在了 / 改指到别人了 —— blocker 留 pending,
                # reconcile_blockers 那边会把它收干净。
                return False, None

            org = await self.db.get(GithubOrg, membership.org_id)
            if org is None or org.deleted_at is not None:
                return False, None
            if org.offboarding_policy == gh_const.OFFBOARDING_POLICY_EXEMPT:
                # 组织被改成了豁免(社区组织)。扫描时它不会入队,但一条
                # 已经入队的 blocker 可能撞上中途改策略 —— 这时**不撤**,
                # 并把 blocker 标成已处置,理由写在 notes 上。
                blocker.resolution = RESO_REVOKED_AUTO
                blocker.resolution_ref = (
                    f"offboarding_exempt:github_org:{membership.id}"
                )
                blocker.notes = (
                    (blocker.notes or "")
                    + f"\n[离职] {org.slug} 已设为离职豁免(社区组织),"
                    f"保留成员资格"
                ).strip()
                return True, None

            # 异常**故意**向上抛到 _force_transfer_all 的 per-blocker
            # SAVEPOINT:它记 last_error_* 并留 pending,下个 tick 重试,
            # triage_bucket 升级为 needs_it。执行层在驱动失败时不会标
            # removed —— 那会在台账上留一条假的回收记录,而那条记录正是
            # 以后所有人判断「这个人还有没有访问权」的依据。
            outcome = await gh_exec.revoke_membership(
                self.db,
                membership=membership,
                org=org,
                reason="offboarding",
                removal_source=gh_const.REMOVAL_SOURCE_OFFBOARDING,
                trace_id=str(process.id),
                actor_person_id=(
                    str(default_new_owner.id) if default_new_owner else None
                ),
            )
            if outcome.outcome == gh_exec.OUTCOME_FAILED:
                raise RuntimeError(
                    f"GitHub org removal failed for "
                    f"{membership.github_login}@{org.slug}: {outcome.error}"
                )
            if outcome.outcome == gh_exec.OUTCOME_MANUAL:
                # 没配凭证 —— **绝不标已回收**:那个人还在组织里。
                #
                # 这里 raise 而不是 ``return False, None``,是因为两者都会
                # 把 blocker 留在 pending(于是 ``_advance_single`` 照样开
                # LEAVER_ORPHAN),但只有 raise 会让 per-blocker SAVEPOINT
                # 把原因记进 ``last_error_*`` 并把 triage_bucket 升成
                # ``needs_it``。静默返回换来的是一个没有任何解释的
                # critical 事件 —— IT 看到「force-transfer 本该处理却没处理」
                # 这个标题,而真实原因是「凭证还没配」。
                #
                # 凭证配好之前每个离职者都会撞到这条,这是**对的**:那些人
                # 确实还持有 GitHub 访问权。让它安静下来,正是这套东西
                # 存在要消灭的那种沉默。
                why = outcome.error or (
                    "no GitHub credential configured for this organization"
                )
                raise RuntimeError(
                    f"cannot remove {membership.github_login} from "
                    f"{org.slug}: {why}"
                )

            blocker.resolution = RESO_REVOKED_AUTO
            blocker.resolution_ref = (
                f"offboarding_revoke:github_org_membership:{membership.id}"
            )
            blocker.notes = (
                (blocker.notes or "")
                + f"\n[离职] 自动移出 GitHub 组织 {org.slug}"
            ).strip()
            return True, None

        if resource_kind == RES_AI_TOOL_APPLICATION:
            from app.domains.ai_tool_application.models import (  # noqa: PLC0415
                AI_TOOL_STATUS_APPROVED,
                AiToolApplication,
            )
            from app.domains.ai_tool_application.service import (  # noqa: PLC0415
                AiToolApplicationService,
            )
            ai_row = await self.db.get(
                AiToolApplication, str(resource_id),
            )
            if (
                ai_row is None
                or ai_row.person_id != str(person.id)
                or ai_row.status != AI_TOOL_STATUS_APPROVED
            ):
                return False, None
            # Exceptions propagate — see VPN branch comment.
            await AiToolApplicationService(self.db).revoke(
                app_id=str(ai_row.id),
                actor=default_new_owner,
                reason="offboarding",
                trace_id=str(process.id),
            )
            blocker.resolution = RESO_REVOKED_AUTO
            blocker.resolution_ref = (
                f"offboarding_revoke:ai_tool_application:{ai_row.id}"
            )
            blocker.notes = (
                (blocker.notes or "")
                + "\n[离职] 自动 revoke AI 工具申请"
            ).strip()
            return True, None

        if resource_kind == RES_CLOUD_ACCESS_GRANT:
            grant = await self.db.get(
                CmdbCloudAccessGrant, resource_id,
            )
            if grant is None or grant.person_id != str(person.id):
                return False, None
            from app.core.rbac import GLOBAL_ADMIN_ROLE  # noqa: PLC0415
            from app.domains.cmdb.lifecycle_service import (  # noqa: PLC0415
                CloudLifecycleService,
            )
            # Exceptions propagate — see VPN branch comment. Note we
            # explicitly DROP ``suppress_driver_error=True`` here: the
            # pre-148 contract was "swallow driver errors so the blocker
            # always marks done"; the new contract is "fail loud so the
            # blocker stays pending and surfaces in needs_it."
            #
            # Round-21 (2026-05-13) prod incident: round-148 dropped
            # ``suppress_driver_error=True`` to gain "fail loud" on driver
            # errors — but ``suppress_driver_error`` was also the implicit
            # "system-context, skip RBAC" signal (see comment block at
            # lifecycle_service.py ~line 2061: "only the grant holder, a
            # cloud admin, the scope owner, or the offboarding sweep
            # (suppress_driver_error=True signals system-context) may
            # revoke"). With both signals dropped the RBAC defense-in-
            # depth rejected every offboarding force-revoke because the
            # actor (manager / fallback owner) is neither the grant
            # holder nor a cloud admin nor the scope owner — raising a
            # 403 AppException per blocker, stacking 14 unresolved
            # cloud_access_grant blockers on huohao@/frank.shen@'s
            # offboarding processes and surfacing as a generic 500 in the
            # admin's "立即推进" click. Restore the system-context signal
            # explicitly by vouching for the actor with global_admin
            # role-names; ``suppress_driver_error`` stays False so cloud-
            # side driver failures still raise (the round-148 contract).
            await CloudLifecycleService(self.db).revoke_grant(
                str(grant.id),
                reason="offboarding",
                actor_user_id=default_new_owner.id,
                trace_id=str(process.id),
                actor_role_names=[GLOBAL_ADMIN_ROLE],
            )
            blocker.resolution = RESO_REVOKED_AUTO
            blocker.resolution_ref = (
                f"offboarding_revoke:{grant.id}"
            )
            blocker.notes = (
                (blocker.notes or "")
                + "\n[离职] 自动 revoke 云访问 grant"
            ).strip()
            return True, None

        if resource_kind == RES_OBJECT_STORAGE:
            # Object-storage bucket owner (2026-09-05, PR3). Same shape as
            # LB / CDN: only Helix's owner FK moves, the bucket itself is
            # untouched. No per-resource owner-history table exists for
            # OSS, so the audit-only recorder is used.
            oss = await self.db.get(CmdbObjectStorage, resource_id)
            if oss is None or oss.owner_person_id != str(person.id):
                return False, None
            previous_owner = oss.owner_person_id
            oss.owner_person_id = str(chosen_owner.id)
            new_owner_label = (
                chosen_owner.work_email or chosen_owner.display_name
            )
            if new_owner_label:
                oss.owner_raw = new_owner_label
            label = blocker.label or oss.bucket_name or str(oss.id)
            await _record_owner_force_transfer_audit(
                resource_type="object_storage",
                resource_id_str=str(oss.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.object_storage.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "object_storage",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_CERTIFICATE:
            # Only MANUALLY-assigned certificate owners are the leaver's to
            # hand over — a probed / cloud-synced certificate resolves its
            # owner live from the domain chain (migration 360), and moving
            # that FK would fight the resolver. The scanner already filters
            # on ``owner_is_manual``; re-check here because the flag can be
            # cleared between the scan and T+0.
            cert = await self.db.get(CmdbCertificate, resource_id)
            if (
                cert is None
                or cert.owner_person_id != str(person.id)
                or not cert.owner_is_manual
            ):
                return False, None
            previous_owner = cert.owner_person_id
            cert.owner_person_id = str(chosen_owner.id)
            label = blocker.label or cert.common_name or str(cert.id)
            await _record_network_transfer(
                resource_type=RESOURCE_TYPE_CERTIFICATE,
                resource_id_str=str(cert.id),
                previous_owner=previous_owner,
                new_owner=str(chosen_owner.id),
                audit_action="cmdb.certificate.owner_force_transferred",
                label=label,
            )
            return True, {
                "type": "certificate",
                "label": label,
                "recipient_id": str(chosen_owner.id),
                "recipient_label": recipient_label,
            }

        if resource_kind == RES_EMAIL_GROUP_MEMBER:
            # Mail-group membership (2026-09-05, PR3): people-bound, so it
            # is REMOVED at the source rather than transferred. Exceptions
            # propagate to the SAVEPOINT handler (same contract as the VPN
            # branch) — a Google refusal must leave the blocker pending and
            # visible, never mark it done.
            from app.domains.email_group.membership_scan import (  # noqa: PLC0415
                remove_person_from_group,
            )
            from app.domains.email_group.models import EmailGroup  # noqa: PLC0415

            group = await self.db.get(EmailGroup, resource_id)
            if group is None or (group.status or "active") != "active":
                return False, None
            await remove_person_from_group(
                self.db,
                group=group,
                person=person,
                trace_id=str(process.id),
                actor_user_id=None,
                reason="offboarding",
            )
            blocker.resolution = RESO_REVOKED_AUTO
            blocker.resolution_ref = (
                f"offboarding_revoke:email_group_member:{group.id}"
            )
            blocker.notes = (
                (blocker.notes or "") + "\n[离职] 自动移出邮件组"
            ).strip()
            return True, None

        if resource_kind == RES_DATA_ACCESS_GRANT:
            # DAX personal dataset grant (migration 606). System-context
            # revoke at the data_access domain; exceptions propagate to the
            # SAVEPOINT handler like the VPN / cloud-grant branches so a
            # failed revoke stays ``pending`` and surfaces in needs_it.
            from app.domains.data_access.lifecycle import (  # noqa: PLC0415
                DataAccessLifecycleService,
            )
            handled = await DataAccessLifecycleService(self.db).revoke_for_offboarding(
                resource_id,
                leaver_person_id=person.id,
                actor_person_id=default_new_owner.id if default_new_owner else None,
                trace_id=str(process.id),
            )
            if not handled:
                return False, None
            blocker.resolution = RESO_REVOKED_AUTO
            blocker.resolution_ref = (
                f"offboarding_revoke:data_access_grant:{resource_id}"
            )
            blocker.notes = (
                (blocker.notes or "")
                + "\n[离职] 自动 revoke DAX 个人数据集授权"
            ).strip()
            return True, None

        logger.warning(
            "force_transfer: unknown resource_type %r on blocker %s",
            resource_kind, blocker.id,
        )
        return False, None

    async def _disable_unclaimed_consumer(
        self,
        blocker: OffboardingBlocker,
        person: IdentityPerson,
        process: OffboardingProcess,
    ) -> bool:
        """Disable the identity-consumer behind ``blocker`` — stop its PII
        egress now that the owner has left and nobody claimed it.

        A data tap must NOT keep running under a departed owner. Disabling
        is a SYSTEM action that needs no successor, so — unlike the
        owner-FK transfers — it must run independently of fallback-person
        resolution and the 24h grace hold. An admin can re-enable +
        reassign via the console if the tap is still needed.

        Returns True if the blocker was resolved (consumer disabled or
        already disabled), False if the consumer no longer matches this
        leaver (stale blocker — caller leaves it pending).
        """
        from app.identity.consumers.models import (  # noqa: PLC0415
            CONSUMER_STATUS_DISABLED,
            IdentityConsumer,
        )

        consumer = await self.db.get(IdentityConsumer, blocker.resource_id)
        if (
            consumer is None
            or consumer.deleted_at is not None
            or consumer.owner_person_id != person.id
        ):
            return False
        if consumer.status != CONSUMER_STATUS_DISABLED:
            consumer.status = CONSUMER_STATUS_DISABLED
        blocker.resolution = RESO_REVOKED_AUTO
        blocker.resolution_ref = (
            f"offboarding_disable:identity_consumer:{consumer.id}"
        )
        blocker.notes = (
            (blocker.notes or "")
            + "\n[离职] 负责人离职且无人接管 → 自动停用身份消费方"
        ).strip()
        from app.audit.service import AuditService as _Audit  # noqa: PLC0415

        await _Audit(self.db).log(
            trace_id=str(process.id),
            actor_type="system",
            actor_user_id=None,
            action="identity_consumer.owner_leaver_auto_disabled",
            status="success",
            resource_type="identity_consumer",
            resource_id=str(consumer.id),
            detail={"slug": consumer.slug,
                    "previous_owner_person_id": str(person.id),
                    "reason": "owner_leaver_unclaimed"},
        )
        return True

    async def _disable_unclaimed_consumers(
        self,
        *,
        person: IdentityPerson,
        process: OffboardingProcess,
        blockers: list[OffboardingBlocker],
    ) -> int:
        """Disable every unclaimed identity-consumer in ``blockers``.

        Runs BEFORE the immediate/grace bucket split so consumer egress is
        cut on the first tick regardless of whether a fallback person can
        be resolved — the owner-FK transfers are gated behind
        ``_resolve_fallback_person``, but a PII data tap must never stay
        live just because the manager chain / IT-admin fallback is empty.
        Per-blocker SAVEPOINT isolation mirrors ``_force_transfer_all``.
        """
        count = 0
        for blocker in blockers:
            if blocker.resolution != RESO_PENDING:
                continue
            try:
                async with self.db.begin_nested():
                    if await self._disable_unclaimed_consumer(
                        blocker, person, process,
                    ):
                        count += 1
            except Exception as exc:  # noqa: BLE001
                if is_transaction_aborting_db_error(exc):
                    raise
                logger.exception(
                    "offboarding: disable unclaimed consumer failed "
                    "(blocker=%s person=%s)",
                    blocker.id, person.id,
                )
        return count

    async def _force_transfer_all(
        self,
        *,
        person: IdentityPerson,
        process: OffboardingProcess,
        blockers: list[OffboardingBlocker],
        new_owner: IdentityPerson,
        emit_sso_bulk_revoke_summary: bool = True,
    ) -> int:
        """Rewrite each resource's owner FK in place + flip blockers.

        We do NOT open ``AssetTransferRequest`` rows and wait for
        accept — that's the whole point of force-transfer. Instead
        we create a synthetic transfer row (status=XFER_COMPLETED)
        for audit parity with the voluntary-transfer path.

        Skipped resource types:
          - ``software_seat`` and ``software_permission`` are
            people-bound (per the scan-side comment "Software has no
            must_transfer — seats are people-bound"). Their offboarding
            path is the auto_revoke provisioning pipeline — transferring
            ownership in our DB without touching the vendor account is
            a no-op that creates a misleading audit trail. Round-13
            (2026-05-08) caught this on P01121: 2 stuck Zoom seats fell
            through to force-transfer and crashed on a missing
            asset_label, then poisoned the session for the hardware
            blocker. Letting auto_revoke own these types fully removes
            that whole class of cascade.

        Per-blocker SAVEPOINT isolation: each iteration is wrapped in
        ``begin_nested()``. A failure on blocker N rolls back only
        blocker N's writes, leaving blockers 1..N-1 committed and
        blocker N+1 free to proceed against a clean session. Pre-round
        13 a single IntegrityError on the first iteration left every
        subsequent blocker dead-lettered with PendingRollbackError.
        """
        count = 0
        now = datetime.now(UTC)
        transferred_items: list[dict[str, Any]] = []
        # Round-59 follow-up (2026-05-31): SSO grants revoked via the
        # direct-revoke branch are collected here so the post-loop
        # fan-out can emit ONE coalesced "[Action Required] Disable
        # SSO for {leaver} — N grants" card per IT admin instead of
        # one card per (grant × admin) pair. Pre-fix: 5 SSO grants × 4
        # admins = 20 admin cards in one tick.
        sso_revoked: list[dict[str, Any]] = []

        for blocker in blockers:
            if blocker.resolution != RESO_PENDING:
                continue
            resource_kind = blocker.resource_type
            resource_id = blocker.resource_id
            if resource_id is None:
                continue
            # Software seats / permissions / accounts: the auto_revoke
            # ProvisioningTask pipeline owns these — see
            # ``SCAN_SCHEDULED_RESOURCE_TYPES`` at module top and the
            # class docstring. Round-58 (2026-05-30) added
            # ``RES_SOFTWARE_ACCOUNT``: ``_apply_account_policy`` writes
            # ProvisioningTask rows for Jira full-access revoke with
            # ``not_before = leaver_date + 1``, but the absence here
            # used to let a still-pending account blocker fall into
            # ``_handle_one_blocker``'s unknown-resource-type tail —
            # emitting a warning, returning (False, None), and then
            # tripping the LEAVER_ORPHAN compliance incident in
            # ``_advance_single`` (line 482). Skipping in the same
            # bucket as seat / permission removes that whole class of
            # spurious incident.
            if resource_kind in SCAN_SCHEDULED_RESOURCE_TYPES:
                continue

            # Round-22 (2026-05-13) hardening: snapshot the mutable
            # attributes that direct-revoke / synthetic-transfer
            # branches mutate INSIDE the savepoint. SQLAlchemy's
            # savepoint rollback reverts DB writes but does NOT
            # revert Python identity-map state, so on exception we
            # restore from this snapshot — see the except clause
            # below for the full failure mode.
            pre_savepoint_snapshot = {
                "resolution": blocker.resolution,
                "resolution_ref": blocker.resolution_ref,
                "notes": blocker.notes,
                "transfer_request_id": blocker.transfer_request_id,
            }
            # Round-59 follow-up (2026-05-31): in-memory rollback of
            # `transferred_items`. `_handle_one_blocker` appends to this
            # list BEFORE the synthetic-transfer INSERT flush, so a flush
            # failure (FK violation, NOT NULL, deadlock, AppException in
            # cloud_access_grant RBAC) leaves a speculative entry on the
            # list. The post-loop notification fan-out (lines ~2588+)
            # then sends "[自动转让] X 项资产已转给你" to the recipient
            # naming a transfer that never committed, plus mirrors the
            # entry into `held_for_it` / `returned_items_by_lender`
            # buckets. On the next 15-minute tick the blocker re-runs
            # (resolution restored from snapshot, error_count++) and the
            # whole speculative-notify cycle repeats. Snapshot the list
            # length pre-savepoint and truncate on except so the post-
            # loop fan-out only sees committed transfers.
            pre_transferred_len = len(transferred_items)
            # Same in-memory rollback for the SSO coalesced summary
            # collector — appended inside `_handle_one_blocker` before
            # the savepoint commits.
            pre_sso_revoked_len = len(sso_revoked)

            try:
                # Per-blocker SAVEPOINT — failure isolated to this
                # iteration only. A flush failure inside this block
                # rolls back to the savepoint; the outer transaction
                # (and other blockers' work) is unaffected.
                async with self.db.begin_nested():
                    moved, transferred_item = await self._handle_one_blocker(
                        blocker=blocker,
                        person=person,
                        process=process,
                        default_new_owner=new_owner,
                        now=now,
                        sso_revoked_collector=sso_revoked,
                    )
                    if transferred_item is not None:
                        transferred_items.append(transferred_item)
                    # Pre-launch P0 fix (synthetic-transfer false-positive):
                    # if the branch didn't actually mutate state (resource
                    # was disposed / revoked / no longer owned by the
                    # leaver between scan time and T+0), skip the synthetic
                    # transfer record — otherwise we'd mark blocker
                    # RESO_TRANSFERRED with a fake xfer pointing at a
                    # transfer-of-nothing.
                    if not moved:
                        logger.info(
                            "force_transfer: blocker %s "
                            "(resource_type=%s, resource_id=%s) found "
                            "resource missing or no longer owned by "
                            "leaver — skipping synthetic transfer; "
                            "blocker stays pending for post_leaver_verify "
                            "to escalate.",
                            blocker.id, resource_kind, resource_id,
                        )
                        # Still mark the provisioning task done if any;
                        # the underlying resource is no longer the
                        # leaver's, so re-attempting a connector call
                        # has nothing to revoke.
                        await self._mark_provisioning_task_done(blocker, now=now)
                        continue
                    # Direct-revoke branches set blocker.resolution
                    # themselves and return ``transferred_item is None,
                    # moved=True`` to signal "counted, no synthetic
                    # transfer needed".
                    if transferred_item is None:
                        await self._mark_provisioning_task_done(blocker, now=now)
                        # Successful retry — clear any previously
                        # recorded failure so the UI doesn't show
                        # stale errors next to a resolved blocker.
                        blocker.last_error_message = None
                        blocker.last_error_at = None
                        count += 1
                        continue

                    # Hardware disposition outcomes that did NOT hand the
                    # asset to a PERSON (held_by_leaver via ``keep`` /
                    # degraded recycle, or office inventory via
                    # ``recycle_to_office``) are NOT force-transfers. The
                    # disposition already wrote the truthful AssetOperation /
                    # AssetHistory rows inside ``_apply_hardware_disposition``.
                    # Minting a synthetic ``AssetTransferRequest`` here would
                    # (a) create a COMPLETED leaver→manager transfer the lead
                    # then sees in /my/todos for a device they never received
                    # (``recipient_id`` is the ``_resolve_hardware_recipient``
                    # fallback, i.e. the manager when the asset's office has
                    # no admin), and (b) stamp the blocker note "owner 改为
                    # {lead}" — both false. Resolve the blocker truthfully and
                    # keep the item ONLY for the correctly-worded notification
                    # fan-out (held_by_leaver → IT; recycle → office admin).
                    disp_action = transferred_item.get("disposition_action")
                    if disp_action in (
                        "keep",
                        "recycle_to_office",
                        "recycle_to_office_degraded",
                    ):
                        if disp_action == "recycle_to_office":
                            note_detail = (
                                "T+0 回收至办公室 "
                                + str(transferred_item.get("recipient_label", ""))
                            )
                        else:
                            note_detail = (
                                "T+0 标记为离职持有 (held_by_leaver),"
                                "资产留在离职员工处,待 IT 追回"
                            )
                        blocker.resolution = RESO_TRANSFERRED
                        blocker.resolution_ref = f"disposition:{disp_action}"
                        blocker.notes = (
                            (blocker.notes or "")
                            + "\n[资产处置] " + note_detail
                        ).strip()
                        # No synthetic transfer_request — leave the pointer
                        # null so the offboarding reconcile loop skips it and
                        # /my/todos shows no phantom incoming transfer.
                        blocker.last_error_message = None
                        blocker.last_error_at = None
                        await self._mark_provisioning_task_done(blocker, now=now)
                        count += 1
                        continue

                    # Synthetic transfer record for audit parity. The
                    # recipient is the per-blocker chosen_owner that the
                    # branch recorded into transferred_item.
                    chosen_owner_id = UUID(transferred_item["recipient_id"])
                    chosen_owner_label = transferred_item.get(
                        "recipient_label", ""
                    )
                    transfer = AssetTransferRequest(
                        initiator_id=person.id,
                        recipient_id=chosen_owner_id,
                        asset_type=resource_kind,
                        asset_ref_id=resource_id,
                        # Round-13 hardening: ``asset_label`` is NOT NULL
                        # on ``asset_transfer_requests``; the previous
                        # missing arg crashed every T+0 force-transfer
                        # for hardware / software_seat / etc on the very
                        # first INSERT and dead-lettered all subsequent
                        # blockers in the same tick with PendingRollback.
                        # Fallback to resource_id string guards stale prod
                        # rows whose label column pre-dates the NOT NULL
                        # migration.
                        asset_label=blocker.label or str(blocker.resource_id),
                        status=XFER_COMPLETED,
                        reason=(
                            f"系统自动强制转让 (T+0): "
                            f"{person.display_name or person.work_email} 离职"
                        ),
                        offboarding_process_id=process.id,
                        completed_at=now,
                        responded_at=now,
                    )
                    self.db.add(transfer)
                    await self.db.flush()

                    blocker.resolution = RESO_TRANSFERRED
                    blocker.transfer_request_id = transfer.id
                    blocker.resolution_ref = f"force_transfer:{transfer.id}"
                    blocker.notes = (
                        (blocker.notes or "")
                        + "\n[系统强制转让] T+0 自动将 owner 改为 "
                        + chosen_owner_label
                    ).strip()
                    # Successful retry — clear stale error fields.
                    blocker.last_error_message = None
                    blocker.last_error_at = None
                    await self._mark_provisioning_task_done(blocker, now=now)
                    count += 1
            except Exception as exc:  # noqa: BLE001
                # A lock-wait (1205) / deadlock (1213) / orphaned-savepoint
                # (1305) aborted the WHOLE transaction server-side — every
                # savepoint, including this blocker's and the outer person's,
                # is gone. Recording it on ``blocker.last_error_message`` and
                # continuing would run the rest of the loop (and the next
                # person) on a dead session, surfacing later as the
                # ``RELEASE SAVEPOINT ... does not exist`` cascade. Re-raise
                # so the whole tick aborts cleanly and the handler retries on
                # a fresh session (the cron is idempotent — it re-scans from
                # live ``leaver_pending`` state).
                if is_transaction_aborting_db_error(exc):
                    logger.warning(
                        "force_transfer: transaction-aborting DB error on "
                        "blocker %s — re-raising for fresh-session retry: %s",
                        blocker.id, exc,
                    )
                    raise
                logger.exception(
                    "force_transfer: failed blocker %s: %s", blocker.id, exc,
                )
                # SAVEPOINT rolled back DB writes. But SQLAlchemy does
                # NOT expire the in-memory identity-mapped attributes
                # set inside the savepoint — so any branch that
                # mutated ``blocker.resolution`` / ``resolution_ref``
                # / ``notes`` / ``transfer_request_id`` BEFORE the
                # failure point (notably the SSO_GRANT direct-revoke
                # branch, which appends to ``notes`` and sets
                # ``resolution_ref`` before any flush; or the synthetic-
                # transfer path that sets ``transfer_request_id`` then
                # raises on the INSERT flush) leaves that speculative
                # state on the Python object. The outer ``self.db.
                # flush()`` would then persist it — silently marking
                # the blocker ``revoked_auto`` with a misleading note
                # line, or pointing ``transfer_request_id`` at a row
                # the savepoint rolled back.
                #
                # Round-22 (2026-05-13) hardening: restore the
                # speculative mutations from ``pre_savepoint_snapshot``
                # captured just before ``begin_nested``. Cannot use
                # ``db.refresh(blocker)`` here because we're still in
                # the per-blocker exception unwind and the async
                # greenlet context for the rolled-back savepoint may
                # not support a fresh IO round-trip — the snapshot
                # approach is pure in-memory and never raises.
                blocker.resolution = pre_savepoint_snapshot["resolution"]
                blocker.resolution_ref = pre_savepoint_snapshot["resolution_ref"]
                blocker.notes = pre_savepoint_snapshot["notes"]
                blocker.transfer_request_id = (
                    pre_savepoint_snapshot["transfer_request_id"]
                )
                # Round-59 follow-up: truncate any speculative entries
                # appended to ``transferred_items`` inside the failed
                # savepoint. Without this the post-loop notification
                # fan-out spuriously pages recipients about transfers
                # that the SAVEPOINT just rolled back, and a 15-minute
                # retry tick repeats the spurious page.
                del transferred_items[pre_transferred_len:]
                # Same truncate for the SSO coalesced collector — a
                # rolled-back SSO revoke must NOT be named in the
                # post-loop "[Action Required] Disable SSO for X" card.
                del sso_revoked[pre_sso_revoked_len:]
                # Record the failure on the structured error columns
                # (migration 148) — OVERWRITE last_error_message rather
                # than append, so a blocker that has bounced 9 times
                # doesn't accumulate 9 multi-KB stack traces in a
                # free-text field. error_count tracks the retry depth;
                # the triage_bucket logic promotes any pending blocker
                # with error_count >= 1 into the ``needs_it`` queue so
                # admins see it. err_summary truncated to 1000 chars
                # (was 500 in the legacy notes-append path) — Pending
                # Rollback / IntegrityError dumps from asyncmy can be
                # very long and the UI tooltip caps render anyway.
                #
                # Round-21 (2026-05-13) prod incident: ``AppException``
                # stores its user-facing text in ``self.message`` and
                # passes nothing to ``Exception.__init__`` — so a naive
                # ``str(exc)`` returns ``""`` and we were silently
                # writing ``"AppException: "`` (15 chars, no signal)
                # for every cloud_access_grant RBAC failure on huohao@
                # / frank.shen@. Use the canonical helper that captures
                # both ``.code`` + ``.message`` for AppException-derived
                # raises and falls back to ``str``/``repr`` for the
                # standard-exception paths (IntegrityError, etc.).
                blocker.last_error_message = format_exception_for_log(
                    exc, max_len=1000,
                )
                blocker.last_error_at = datetime.now(UTC)
                blocker.error_count = (blocker.error_count or 0) + 1
                continue

        await self.db.flush()

        # ── Round-59 follow-up (2026-05-31): SSO coalesced summary ──
        # `SsoGrantService.revoke` was called with
        # ``notify_admins=False`` for every SSO grant inside the per-
        # blocker savepoint, so IT admins received NO per-grant card.
        # Emit ONE coalesced "[Action Required] Disable SSO for X — N
        # grants" card per active IT admin listing every grant that
        # actually committed this tick (entries appended INSIDE the
        # savepoint and rolled back on failure).
        #
        # ``emit_sso_bulk_revoke_summary=False`` is passed by the caller
        # when ``_notify_it_grace_window`` is about to fire in the same
        # tick — the grace card already enumerates immediate-revoked SSO
        # grants in its ``immediate_revoked_lines`` section, so the
        # bulk-revoke card would be a redundant duplicate.
        if sso_revoked and emit_sso_bulk_revoke_summary:
            try:
                from app.scheduled_jobs.failure_notifier import (  # noqa: PLC0415
                    list_admin_recipients,
                )

                admins = await list_admin_recipients(self.db)
                if admins:
                    leaver_label = (
                        person.display_name
                        or person.work_email
                        or str(person.id)
                    )
                    grant_list = "\n".join(
                        f"  · {g['system_name']} (grant {g['grant_id']})"
                        for g in sso_revoked
                    )
                    body = (
                        f"员工 {leaver_label} 今日离职 (T+0),Helix 已自动"
                        f"将以下 {len(sso_revoked)} 项 SSO 授权 revoke。"
                        "SSO 在 Helix 是 DB-only 翻状态,真正的 SAML/SSO "
                        "登录权限仍在第三方控制台 (Google admin / Atlassian"
                        " / Okta 等),请在对应控制台手动关闭:\n\n"
                        f"{grant_list}\n\n"
                        "— Helix IT 平台"
                    )
                    for admin in admins:
                        try:
                            await self.notify.dispatch(
                                NotificationEnvelope(
                                    recipient=admin,
                                    title=(
                                        f"[Action Required] Disable SSO "
                                        f"for {leaver_label} — "
                                        f"{len(sso_revoked)} grants"
                                    ),
                                    body_text=body,
                                    channels=["feishu", "email"],
                                    template_key=(
                                        "offboarding.t0_sso_bulk_revoke"
                                    ),
                                    # Admins land on the leaver's offboarding
                                    # page for context (the actual SAML
                                    # disable is external). Without ``url``
                                    # the offboarding_process fallback sent
                                    # admins to /my/lifecycle — their own,
                                    # empty, wrong page.
                                    url=f"/admin/lifecycle/person/{person.id}",
                                    context={
                                        "leaver_name": leaver_label,
                                        "leaver_email": person.work_email,
                                        "count": len(sso_revoked),
                                        "grant_list": grant_list,
                                    },
                                    related_resource_type="offboarding_process",
                                    related_resource_id=process.id,
                                ),
                                trace_id="process_advance",
                            )
                        except Exception as exc:  # noqa: BLE001
                            logger.warning(
                                "notify t0_sso_bulk_revoke to admin "
                                "%s failed: %s", admin.id, exc,
                            )
                else:
                    logger.warning(
                        "process_advance: %d SSO grant(s) revoked "
                        "for leaver %s but no eligible IT admin "
                        "recipient — coalesced card skipped (the "
                        "per-system owner cards inside revoke() "
                        "still fired).",
                        len(sso_revoked), person.work_email,
                    )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "notify t0_sso_bulk_revoke block failed: %s", exc,
                )

        # Pre-launch hardening 2026-05-06 (round 3): split
        # ``transferred_items`` into two notification streams.
        #
        # 1. Items where the leaver was the BORROWER of someone else's
        #    asset are auto-returned to the original lender (NOT
        #    transferred to ``new_owner``). The new_owner notification
        #    must NOT include them — telling the successor "X assets
        #    transferred to you" when 3 of those 5 went back to a
        #    different person is misleading and creates downstream
        #    confusion ("where's the laptop you said I got?").
        #
        # 2. Original lenders deserve a proactive "your borrowed asset
        #    has come back" push so they don't have to discover the
        #    return on their next /me visit.
        #
        # Round-13 (2026-05-08): hardware can route to the LOCAL OFFICE
        # ADMIN (Sunnyvale leaver → Kristine Ye, not the Beijing manager
        # 唐小丽). Items are grouped by ``recipient_id`` so each actual
        # new owner gets exactly one push naming exactly the items
        # they received — never mixing across recipients.
        items_by_recipient: dict[str, list[dict[str, Any]]] = {}
        returned_items_by_lender: dict[str, list[dict[str, Any]]] = {}
        recycled_by_office_admin: dict[str, list[dict[str, Any]]] = {}
        for i in transferred_items:
            if i.get("borrow_returned_to_owner"):
                lender_id = i.get("original_lender_id")
                if lender_id:
                    returned_items_by_lender.setdefault(
                        lender_id, [],
                    ).append(i)
                continue
            # Hardware disposition outcomes that did NOT move the asset into
            # the recipient's PERSONAL ownership must never ride the
            # "[自动转让] … 已转给你 (无需接受)" push — that wording tells the
            # manager/lead they now own a device they don't. The leaver's
            # hardware either stays with the leaver or goes to office
            # inventory; it is never silently force-transferred to the lead.
            #   * ``keep`` / ``recycle_to_office_degraded`` → asset stayed
            #     with the leaver (held_by_leaver). Surfaced to IT via the
            #     held-by-leaver notice below — NOT to the manager fallback
            #     that ``_resolve_hardware_recipient`` resolves to when the
            #     asset's office has no admin (the exact path that leaked
            #     held/recycled laptops into a lead's "已转给你" card).
            #   * ``recycle_to_office`` → asset went to OFFICE inventory,
            #     not the office admin's personal name. Gets its own
            #     "已回收至办公室" push (below) so the wording matches reality.
            # Everything else — non-hardware owner rewrites (no
            # ``disposition_action``) plus hardware ``transfer_to_person`` —
            # is a real personal transfer and belongs in force_transfer.
            action = i.get("disposition_action")
            if action in ("keep", "recycle_to_office_degraded"):
                continue
            rid = i.get("recipient_id") or str(new_owner.id)
            if action == "recycle_to_office":
                recycled_by_office_admin.setdefault(rid, []).append(i)
                continue
            items_by_recipient.setdefault(rid, []).append(i)

        # Notify each recipient — one push per recipient with exactly
        # the items routed to them.
        for recipient_id, items in items_by_recipient.items():
            try:
                recipient_obj = await self.db.get(
                    IdentityPerson, recipient_id,
                )
                if recipient_obj is None or not recipient_obj.work_email:
                    continue
                asset_list = "\n".join(
                    f"  · [{i['type']}] {i['label']}" for i in items
                )
                # Recipient may be a plain employee (manager fallback,
                # default IT contact, …) — ``/admin/lifecycle/person/…``
                # is gated by the ``global_admin``/``auditor`` RoleRoute
                # so a non-admin recipient gets 403. ``/my/todos`` is
                # the canonical "what landed on me to act on" hub for
                # every role; admin recipients can navigate further if
                # they need the leaver-side view.
                portal_path = "/my/todos"
                await self.notify.dispatch(
                    NotificationEnvelope(
                        recipient=recipient_obj,
                        title=(
                            f"[自动转让] "
                            f"{person.display_name or person.work_email} 的 "
                            f"{len(items)} 项资产已转给你"
                        ),
                        body_text=(
                            f"员工 {person.display_name or person.work_email} "
                            "今日离职,以下资源已系统自动转到你名下"
                            "(无需接受):\n\n"
                            f"{asset_list}\n\n"
                            f"请到 Helix 平台确认或再转让:\n{portal_path}\n\n"
                            "— Helix IT 平台"
                        ),
                        channels=["feishu", "email"],
                        template_key="force_transfer_received",
                        # ``/my/todos`` is the role-agnostic "what landed on
                        # me" hub the body already points at; the
                        # offboarding_process fallback would instead drop the
                        # recipient on /my/lifecycle (the leaver flow), not
                        # the transferred assets.
                        url=portal_path,
                        context={
                            "recipient_name": (
                                recipient_obj.display_name
                                or recipient_obj.work_email
                            ),
                            "leaver_name": (
                                person.display_name or person.work_email
                            ),
                            "count": len(items),
                            "asset_list": asset_list,
                        },
                        # NB: ``portal_url`` deliberately NOT pinned to
                        # ``portal_path`` here. A caller-supplied portal_url
                        # wins over the renderer's absolute URL injection, so
                        # pinning the relative path made the email CTA line
                        # read "请到 Helix 查看: /my/todos" — unclickable. The
                        # renderer now resolves the absolute URL from the
                        # ``url=portal_path`` set above.
                        related_resource_type="offboarding_process",
                        related_resource_id=process.id,
                    ),
                    trace_id="process_advance",
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "notify force-transfer recipient %s failed: %s",
                    recipient_id, exc,
                )

        # Assets recycled to a physical office at T+0. The office admin is
        # the recipient, but the asset is OFFICE inventory, not their
        # personal device — so this carries accurate "已回收至办公室,请入库"
        # wording instead of the personal-transfer "已转给你" template.
        for recipient_id, items in recycled_by_office_admin.items():
            try:
                recipient_obj = await self.db.get(
                    IdentityPerson, recipient_id,
                )
                if recipient_obj is None or not recipient_obj.work_email:
                    continue
                asset_list = "\n".join(
                    f"  · [{i['type']}] {i['label']}" for i in items
                )
                await self.notify.dispatch(
                    NotificationEnvelope(
                        recipient=recipient_obj,
                        title=(
                            f"[资产回收] "
                            f"{person.display_name or person.work_email} 的 "
                            f"{len(items)} 项资产已回收至办公室"
                        ),
                        body_text=(
                            f"员工 {person.display_name or person.work_email} "
                            "今日离职,以下资产已系统回收至办公室,"
                            "请安排入库 / 盘点:\n\n"
                            f"{asset_list}\n\n"
                            "在 Helix 资产列表可查看并处理:\n/admin/assets/list\n\n"
                            "— Helix IT 平台"
                        ),
                        channels=["feishu", "email"],
                        template_key="offboarding.recycled_to_office",
                        url="/admin/assets/list",
                        context={
                            "recipient_name": (
                                recipient_obj.display_name
                                or recipient_obj.work_email
                            ),
                            "leaver_name": (
                                person.display_name or person.work_email
                            ),
                            "count": len(items),
                            "asset_list": asset_list,
                        },
                        related_resource_type="offboarding_process",
                        related_resource_id=process.id,
                    ),
                    trace_id="process_advance",
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "notify recycle-to-office recipient %s failed: %s",
                    recipient_id, exc,
                )

        # Notify each original lender about their returned asset(s).
        # Failure to look up or notify a lender does NOT block the
        # tick — log and continue; the asset is already restored, only
        # the proactive ping is best-effort.
        for lender_id, items in returned_items_by_lender.items():
            try:
                lender = await self.db.get(IdentityPerson, lender_id)
                if lender is None or not lender.work_email:
                    continue
                if lender.employment_status == "left":
                    # Lender has also been offboarded — no point pinging.
                    continue
                lender_asset_list = "\n".join(
                    f"  · [{i['type']}] {i['label']}" for i in items
                )
                await self.notify.dispatch(
                    NotificationEnvelope(
                        recipient=lender,
                        title=(
                            f"[借用归还] {person.display_name or person.work_email} "
                            f"离职,你出借的 {len(items)} 项资产已自动归还"
                        ),
                        body_text=(
                            f"员工 {person.display_name or person.work_email} "
                            f"今日离职,以下他/她借用你的资产已系统自动归还到你名下:\n\n"
                            f"{lender_asset_list}\n\n"
                            "你可以在 Helix /me 页面看到这些资产。\n\n"
                            "— Helix IT 平台"
                        ),
                        channels=["feishu", "email"],
                        template_key="borrow_auto_returned",
                        context={
                            "recipient_name": lender.display_name or lender.work_email,
                            "leaver_name": person.display_name or person.work_email,
                            "count": len(items),
                            "asset_list": lender_asset_list,
                        },
                        related_resource_type="offboarding_process",
                        related_resource_id=process.id,
                    ),
                    trace_id="process_advance",
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "notify borrow-auto-return to lender %s failed: %s",
                    lender_id, exc,
                )

        # Same-day IT notification for assets that ended up
        # ``held_by_leaver`` at T+0 in a way that needs IT follow-up:
        #   * ``keep`` via the *default* path (no admin disposition picked,
        #     default kicked in) — nobody decided, leaver kept the device.
        #   * ``recycle_to_office_degraded`` — recycle was intended but no
        #     office could be resolved, so it fell back to held_by_leaver.
        # Both leave the device on the leaver and demand an IT recovery, so
        # neither may ride the manager "已转给你" push (excluded above) and
        # both surface here instead. Intentional ``keep`` chosen by an admin
        # is NOT surfaced (admin already knew). Distinct from the weekly
        # ``asset.held_by_leaver_reminder`` digest (any-cause); this is the
        # immediate T+0 FYI.
        held_for_it = [
            i for i in transferred_items
            if (
                i.get("disposition_action") == "keep"
                and i.get("default_fallback") is True
            )
            or i.get("disposition_action") == "recycle_to_office_degraded"
        ]
        if held_for_it:
            try:
                from app.scheduled_jobs.failure_notifier import (  # noqa: PLC0415
                    list_admin_recipients,
                )

                admins = await list_admin_recipients(self.db)
                if admins:
                    kept_list = "\n".join(
                        f"  · {i.get('asset_tag') or '—'} — {i['label']}"
                        for i in held_for_it
                    )
                    leaver_label = (
                        person.display_name or person.work_email or str(person.id)
                    )
                    body = (
                        f"员工 [b]{leaver_label}[/b] 今日 T+0,以下 "
                        f"{len(held_for_it)} 项资产已落入「离职持有」状态"
                        "(离职前 IT 与员工均未处置走默认兜底,或回收办公室"
                        "失败降级),仍留在该员工手上,不阻塞离职流程:\n\n"
                        f"{kept_list}\n\n"
                        "处理建议:确认归还/回购后,在资产详情页点「追回」"
                        "把状态翻回闲置;或者在 /admin/lifecycle/person/… "
                        "为该离职流程补一个 disposition。\n\n"
                        "— Helix IT 平台"
                    )
                    for admin in admins:
                        try:
                            await self.notify.dispatch(
                                NotificationEnvelope(
                                    recipient=admin,
                                    title=(
                                        f"[T+0 离职持有] {leaver_label} 的 "
                                        f"{len(held_for_it)} 项资产标记为离职持有待追回"
                                    ),
                                    body_text=body,
                                    channels=["feishu", "email"],
                                    template_key="offboarding.t0_default_held_by_leaver",
                                    # Admins recover held-by-leaver assets on
                                    # the asset list ("追回"); the
                                    # offboarding_process fallback would send
                                    # them to /my/lifecycle instead.
                                    url="/admin/assets/list",
                                    context={
                                        "leaver_name": leaver_label,
                                        "leaver_email": person.work_email,
                                        "count": len(held_for_it),
                                        "asset_list": kept_list,
                                    },
                                    related_resource_type="offboarding_process",
                                    related_resource_id=process.id,
                                ),
                                trace_id="process_advance",
                            )
                        except Exception as exc:  # noqa: BLE001
                            logger.warning(
                                "notify t0_default_held_by_leaver to admin "
                                "%s failed: %s", admin.id, exc,
                            )
                else:
                    logger.warning(
                        "process_advance: %d asset(s) defaulted to "
                        "held_by_leaver for leaver %s but no eligible IT "
                        "admin recipient — skipping same-day push "
                        "(weekly reminder will still cover).",
                        len(held_for_it), person.work_email,
                    )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "notify t0_default_held_by_leaver block failed: %s",
                    exc,
                )

        return count


__all__ = ["ProcessAdvanceService"]