"""Offboarding and asset transfer endpoints."""
from uuid import UUID

from fastapi import APIRouter, Depends, Header, Query, Request
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import get_current_user, require_admin, require_lifecycle_read
from app.core.dependencies import (  # noqa: E402
    require_active_employee_for_mutations as _leaver_runtime_gate,
)
from app.core.idempotency import IdempotencyService
from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role
from app.core.response import paginated_response, success_response
from app.domains.asset.models import AssetItem
from app.domains.offboarding.schemas import (
    AssetTransferCreateRequest,
    AssetTransferResponseBody,
    BlockerHandoverRequest,
    BlockerRevokeRequest,
    BlockerSkipRequest,
    CancelProcessRequest,
    DefaultHandoverPersonRequest,
    FallbackHandoverPersonRequest,
    FallbackHandoverPersonResponse,
    HardwareDispositionPreviewEntry,
    HardwareDispositionPreviewResponse,
    HardwareDispositionRequest,
    HardwareDispositionResponse,
    MyHandoverItem,
    OffboardingPreviewResponse,
    ReassignHandoverRequest,
    SuspendSsoResultResponse,
)
from app.domains.offboarding.service import OffboardingService
from app.domains.offboarding.transferable_summary_service import (
    TransferableSummaryService,
)
from app.domains.offboarding.transferable_summary_service import (
    serialize as _serialize_transferable_summary,
)

router = APIRouter(
    prefix="/api/v1/offboarding",
    tags=["offboarding"],
    dependencies=[Depends(_leaver_runtime_gate)],
)


# ─── Processes (admin) ───


@router.get("/transferable-summary")
async def get_transferable_summary(
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Counts the live transferable-ownership universe per category.

    Powers the "必须转让" tally on ``/admin/offboarding``.
    Six categories — software (must_transfer policy) / cloud scope /
    domain / email group / asset item / IAM user — disjoint by
    construction so the totals add up.

    Read-only and admin-gated; no mutation, no audit row (consistent
    with other dashboard summaries — write paths log, read paths
    don't).
    """
    summary = await TransferableSummaryService(db).get_summary()
    return success_response(
        data=_serialize_transferable_summary(summary),
        request=request,
    )


@router.get("/processes")
async def list_processes(
    request: Request,
    status: str | None = Query(None),
    page: int = Query(1, ge=1),
    # 2026-05-31 (PR-B cockpit fix): bumped le from 100 → 200 so the
    # cockpit's single-shot full-in-flight fetch (~few dozen rows in
    # steady state, headroom for HRIS bulk-confirm bursts) no longer
    # 422s. Page is still an indexed scan + batched bucket counts.
    page_size: int = Query(20, ge=1, le=200),
    # 2026-05-31 (PR-B cockpit fix): cockpit needs per-process blockers
    # for its resource-type filter dropdown and expandable rows. Default
    # stays False so the legacy OffboardingAdminPage's lightweight list
    # call is unchanged; cockpit opts in.
    include_blockers: bool = Query(False),
    search: str | None = Query(None, max_length=100),
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    items, total = await svc.list_processes(
        status=status,
        page=page,
        page_size=page_size,
        include_blockers=include_blockers,
        search=search,
    )
    return paginated_response(
        data=items, total=total, page=page, page_size=page_size, request=request,
    )


# NOTE: must stay registered BEFORE ``GET /processes/{process_id}`` —
# FastAPI matches in registration order and the path-param route would
# otherwise swallow ``export.xlsx`` as a (non-UUID) process id.
@router.get("/processes/export.xlsx")
async def export_processes(
    request: Request,
    status: str | None = Query(None),
    current_user: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Export the filtered offboarding process list as XLSX.

    Filter signature mirrors :func:`list_processes` minus pagination
    (``include_blockers`` is pinned False — the nested blocker payload
    doesn't flatten into a row; the scalar triage counts do). Reuses
    ``OffboardingService.list_processes`` so any filter/RBAC fix on the
    list applies here automatically. Capped at
    :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows.
    """
    from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response

    svc = OffboardingService(db)
    items, total = await svc.list_processes(
        status=status,
        page=1,
        page_size=EXPORT_ROW_LIMIT,
        include_blockers=False,
    )
    columns = [
        ("person_name", "人员"),
        ("person_email", "邮箱"),
        ("department", "部门"),
        ("status", "状态"),
        ("leaver_date", "最后工作日"),
        ("sla_breach_level", "SLA等级"),
        ("pending_blocker_count", "待处理阻塞项"),
        ("needs_it_count", "需IT介入"),
        ("auto_in_progress_count", "系统处理中"),
        ("resolved_count", "已解决"),
        ("total_blocker_count", "阻塞项总数"),
        ("default_handover_person_name", "默认接手人"),
        ("sso_suspended_at", "SSO停用时间"),
        ("ready_at", "就绪时间"),
        ("completed_at", "完成时间"),
        ("created_at", "创建时间"),
        ("id", "流程ID"),
    ]
    rows = [p.model_dump(exclude={"blockers"}) for p in items]
    return await xlsx_export_response(
        db,
        request,
        current_user,
        columns=columns,
        rows=rows,
        total_matched=total,
        filename_stem="offboarding_processes",
        sheet_title="Offboarding Processes",
        audit_action="offboarding.processes.export",
        audit_resource_type="offboarding_process",
        filters={"status": status},
    )


@router.get("/analytics")
async def offboarding_analytics(
    request: Request,
    window_days: int = Query(30, ge=1, le=90),
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Single-trip analytics aggregate for the offboarding-admin charts.

    Admin-read gated. Returns process KPI + status/SLA rollups + live
    blocker distributions (resource_type / severity) + daily-leavers
    trend. Read-only augment; no MCP tool (mirrors the existing
    定时任务/审计/审批 analytics tabs).
    """
    svc = OffboardingService(db)
    data = await svc.analytics_summary(window_days=window_days)
    return success_response(data=data, request=request)


@router.get("/my-team-leavers")
async def list_my_team_leavers(
    request: Request,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List ongoing offboarding processes the current user is
    accountable for as a NON-admin stakeholder.

    Returns leavers whose ``hrbp_person_id`` is the current user
    (URGENT/BREACH SLA tier), whose direct ``manager_person_id`` is the
    current user (WARN SLA tier), or whose skip-level manager is the
    current user. Backs ``/my/team-leavers``, reached from the Feishu
    SLA card's "Handle in Helix" button when the recipient is not in
    the ``global_admin`` pool. Completed and cancelled processes are
    filtered out.
    """
    svc = OffboardingService(db)
    items, total = await svc.list_my_team_leavers(
        current_user.id, page=page, page_size=page_size,
    )
    return paginated_response(
        data=items, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/my-handovers", summary="Handovers the current user holds")
async def list_my_handovers(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Offboarding processes where the current user is the handover person.

    Self-service (any authenticated employee): the handover person is
    usually picked by rule — the leaver's lead — and is frequently NOT the
    colleague who actually took over the work. This is the surface that
    lets them see what is routing to them and hand it on themselves,
    instead of the data landing wherever the org chart pointed.

    Only handovers that can still change something are returned (open
    processes, or completed ones whose vendor deletes are still pending).
    """
    svc = OffboardingService(db)
    items = await svc.list_my_handovers(current_user.id)
    return success_response(
        data=[MyHandoverItem(**i).model_dump(mode="json") for i in items],
        request=request,
    )


@router.put(
    "/my-handovers/{process_id}/reassign",
    summary="Hand a handover you hold to a colleague",
)
async def reassign_my_handover(
    request: Request,
    process_id: UUID,
    body: ReassignHandoverRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Reassign a handover to a colleague.

    Authorisation is enforced in the service: only the CURRENT handover
    person (or a global_admin) may reassign. Without that check, any
    authenticated employee could redirect a departing colleague's Drive
    content to themselves — this route is intentionally not admin-gated,
    so the ownership check IS the gate.
    """
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    is_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE)
    result = await svc.reassign_my_handover(
        process_id,
        current_user_id=current_user.id,
        new_person_id=body.new_person_id,
        trace_id=trace_id,
        is_admin=is_admin,
    )
    await db.commit()
    return success_response(
        data={k: str(v) if v is not None else None for k, v in result.items()},
        request=request,
    )


@router.get("/processes/{process_id}")
async def get_process(
    request: Request,
    process_id: UUID,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    item = await svc.get_process_detail(process_id)
    return success_response(data=item, request=request)


@router.put("/processes/{process_id}/default-handover")
async def set_default_handover_person(
    request: Request,
    process_id: UUID,
    body: DefaultHandoverPersonRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Set or clear the per-process default handover successor (migration 211).

    The default prefills the per-blocker "指定接手人" modal (SSO grant /
    domain registration / google_orphan_owner). NULL clears it. A
    non-NULL value must be an active, non-deleted employee — the
    service-layer validation mirrors the per-blocker handover path so a
    stale default never silently fails downstream.
    """
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    await svc.set_default_handover_person(
        process_id,
        successor_person_id=body.successor_person_id,
        trace_id=trace_id,
        actor_user_id=current_user.id,
    )
    await db.commit()
    detail = await svc.get_process_detail(process_id)
    return success_response(data=detail, request=request)


@router.get(
    "/settings/fallback-handover",
    summary="Read the org-wide fallback handover person",
    response_model=None,
)
async def get_fallback_handover_person(
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Resolved view of ``offboarding.default_handover_fallback_person_id``.

    Last tier of the terminal-disposition successor chain (per-process
    default → leaver's manager → manager chain → this). Returns the
    person's name/email — not the bare UUID — so the admin surface can
    render something readable, plus an ``is_stale`` flag when the
    configured person has since left (the chain then behaves as if
    nothing were configured, and IT is paged instead of data being
    deleted).
    """
    svc = OffboardingService(db)
    data = await svc.get_fallback_handover_person()
    return success_response(
        data=FallbackHandoverPersonResponse(**data).model_dump(mode="json"),
        request=request,
    )


@router.put(
    "/settings/fallback-handover",
    summary="Set the org-wide fallback handover person",
    response_model=None,
)
async def set_fallback_handover_person(
    request: Request,
    body: FallbackHandoverPersonRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Set or clear the org-wide fallback handover person.

    ``global_admin`` only (it decides where *any* orphaned leaver's Drive
    / Calendar / Zoom content lands, org-wide) — deliberately stricter
    than the per-process default, which lifecycle admins manage.
    """
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    data = await svc.set_fallback_handover_person(
        person_id=body.person_id,
        trace_id=trace_id,
        actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(
        data=FallbackHandoverPersonResponse(**data).model_dump(mode="json"),
        request=request,
    )


@router.put("/processes/{process_id}/hardware-disposition")
async def set_hardware_disposition(
    request: Request,
    process_id: UUID,
    body: HardwareDispositionRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Set the per-asset hardware disposition map (Admin only).

    Driven by the offboarding-console UI before T+0 grace expires.
    Each map entry tells the T+0 advance handler what to do with one
    of the leaver's hardware blockers:

      * ``keep`` — leaver retains the device (`held_by_leaver`)
      * ``recycle_to_office`` — return to a city office (default if
        no entry; ``office_id`` may be omitted, in which case the
        handler resolves the leaver's home office)
      * ``transfer_to_person`` — hand to a named successor
        (``person_id`` required)

    Replaces the entire map (PUT semantics). Entries for asset IDs
    that aren't current hardware blockers on this process are
    silently kept on the row (forward-compat with batched template
    workflows).
    """
    svc = OffboardingService(db)
    proc = await svc.set_hardware_disposition(
        process_id=process_id,
        dispositions=body.dispositions,
        actor_user_id=current_user.id,
        trace_id=(
            str(request.state.trace_id)
            if hasattr(request.state, "trace_id") else ""
        ),
    )
    await db.commit()
    return success_response(
        data=HardwareDispositionResponse(
            dispositions=proc.hardware_disposition or {},
            updated_at=proc.updated_at,
        ),
        request=request,
    )


@router.get("/processes/{process_id}/hardware-disposition-preview")
async def hardware_disposition_preview(
    request: Request,
    process_id: UUID,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Suggested default disposition per hardware blocker (Admin only).

    Drives the disposition-picker UI: the admin sees each hardware
    blocker with a pre-filled default (typically
    ``recycle_to_office`` to the leaver's resolved home office) and
    can confirm in bulk or edit individual rows.
    """
    svc = OffboardingService(db)
    entries = await svc.preview_hardware_disposition(process_id)
    return success_response(
        data=HardwareDispositionPreviewResponse(
            entries=[
                HardwareDispositionPreviewEntry(**e) for e in entries
            ],
        ),
        request=request,
    )


@router.post("/processes/{process_id}/reconcile")
async def reconcile_process(
    request: Request,
    process_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    proc = await svc.reconcile_blockers(
        process_id, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=await svc.get_process_detail(proc.id), request=request)


@router.post("/processes/{process_id}/suspend-sso")
async def suspend_sso(
    request: Request,
    process_id: UUID,
    idempotency_key: str | None = Header(
        None, alias="Idempotency-Key", max_length=128),
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Execute every still-pending revoke task for this leaver and stamp
    sso_suspended_at. Returns per-task outcome so the operator can see
    which tasks failed and need a retry from the provisioning page
    (P0-2 surface, 2026-05-08).

    Idempotency: scoped per ``process_id``. Replay returns the cached
    body so a double-click can't fire revoke tasks twice. The service
    is internally idempotent on ``sso_suspended_at`` but the
    provisioning task fan-out is not — this guard protects the
    fan-out.
    """
    idem = IdempotencyService(
        db,
        scope=f"offboarding.suspend_sso:{process_id}",
        key=idempotency_key,
    )
    cached = await idem.lookup()
    if cached is not None:
        return cached

    svc = OffboardingService(db)
    trace_id = (
        str(request.state.trace_id)
        if hasattr(request.state, "trace_id") else ""
    )
    try:
        proc = await svc.suspend_sso(
            process_id, trace_id=trace_id, actor_user_id=current_user.id,
        )
        await db.commit()
    except Exception:
        await idem.mark_failed()
        raise
    suspend_result = getattr(proc, "_suspend_sso_result", None) or {
        "executed": 0, "executed_task_ids": [], "failed_task_ids": [],
    }
    detail = await svc.get_process_detail(process_id)
    payload = SuspendSsoResultResponse(
        process=detail,
        executed_count=suspend_result["executed"],
        executed_task_ids=suspend_result["executed_task_ids"],
        failed_task_ids=suspend_result["failed_task_ids"],
    )
    response = success_response(data=payload, request=request)
    await idem.record(response, resource_id=process_id)
    return response


@router.post("/processes/{process_id}/reenforce-session")
async def reenforce_session(
    request: Request,
    process_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Retry the identity-plane session/account cutoff (Google signOut +
    suspend, Feishu) and re-record per-channel enforcement proof. For an
    operator to re-run a channel that previously failed (e.g. the historical
    ``users.signOut`` scope 403). Idempotent; returns the refreshed process
    detail so the caller sees the updated ``session_enforcement``."""
    svc = OffboardingService(db)
    trace_id = (
        str(request.state.trace_id)
        if hasattr(request.state, "trace_id") else ""
    )
    await svc.reenforce_session_cutoff(
        process_id, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    detail = await svc.get_process_detail(process_id)
    return success_response(data=detail, request=request)


@router.get(
    "/people/{person_id}/preview",
    response_model=None,
    summary=(
        "Read-only dry-run scan of resources an offboarding would touch"
    ),
)
async def get_offboarding_preview(
    request: Request,
    person_id: UUID,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Mirror of ``_rescan`` SELECTs without persisting any blocker /
    provisioning task / audit row. Powers the confirm modal on
    ``/admin/users/{id}`` so the operator sees exactly what
    ``mark-leaver`` would scan, including the auto-picked successor for
    each seat that needs one.

    D-1 fix (2026-05-08).
    """
    svc = OffboardingService(db)
    data = await svc.get_offboarding_preview(person_id)
    # Run the dict through the Pydantic model for shape validation.
    # Mismatch between the service contract and the schema is a bug we
    # want surfaced loudly — better here than as a frontend type error.
    payload = OffboardingPreviewResponse.model_validate(data)
    return success_response(data=payload, request=request)


@router.post("/processes/{process_id}/re-notify")
async def re_notify_process(
    request: Request,
    process_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Admin operator manual retry of dispatch_post_commit_actions.

    Round 2 P2 fix (LG2-005): the HRIS path's post-commit dispatch
    has no automatic retry on failure (a partial Feishu/email outage
    when the leaver / manager start-of-cycle notifications would
    have fired). The escalation cron only handles T-7/T-3/T-1 +
    REMINDER_ESCALATION; SLA service handles warn/urgent/breach.
    Process-started notifications were a permanent-loss class on
    dispatcher exception. This endpoint fills that gap — operator
    sees "post_commit_commit_error" in the mark-leaver response and
    can hit Re-notify from the offboarding console.

    Idempotent via OffboardingReminderLog rows keyed by
    (process_id, kind) — a successful prior dispatch will no-op.
    """
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    # Look up person_id directly so we don't double-call get_process_detail
    # (which loads blockers/reminders we don't need here).
    from app.domains.offboarding.models import OffboardingProcess  # noqa: PLC0415
    proc_row = await db.get(OffboardingProcess, process_id)
    if proc_row is None:
        from app.core.exceptions import NotFoundError  # noqa: PLC0415
        raise NotFoundError("OffboardingProcess", str(process_id))
    summary = await svc.dispatch_post_commit_actions(
        proc_row.person_id,
        process_id,
        trace_id=trace_id,
        actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=summary, request=request)


@router.post("/processes/{process_id}/complete")
async def complete_process(
    request: Request,
    process_id: UUID,
    idempotency_key: str | None = Header(
        None, alias="Idempotency-Key", max_length=128),
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Mark the offboarding process complete.

    Idempotency: scoped per ``process_id`` so a duplicate click can't
    fire the completion-side audit + downstream notifications twice.
    """
    idem = IdempotencyService(
        db,
        scope=f"offboarding.complete:{process_id}",
        key=idempotency_key,
    )
    cached = await idem.lookup()
    if cached is not None:
        return cached

    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    try:
        await svc.complete_process(
            process_id, trace_id=trace_id, actor_user_id=current_user.id,
        )
        await db.commit()
    except Exception:
        await idem.mark_failed()
        raise
    response = success_response(data=await svc.get_process_detail(process_id), request=request)
    await idem.record(response, resource_id=process_id)
    return response


@router.post("/processes/{process_id}/cancel")
async def cancel_process(
    request: Request,
    process_id: UUID,
    body: CancelProcessRequest,
    # Capped so the namespaced composite key (key + ":" + 36-char UUID) can
    # never exceed the 128-char core_idempotency_records.key column — an
    # overlong header otherwise raises 1406 on TiDB and 500s every retry.
    idempotency_key: str | None = Header(
        None, alias="Idempotency-Key", max_length=80),
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Cancel an offboarding process (false-leaver rollback).

    Idempotency: scoped per ``process_id`` so a double-click can't fire
    the cascading auto-restore of cancelled provisioning tasks, the
    asset-transfer cancellations and the seat re-grant enqueue twice.
    The mutation is internally idempotent on
    ``proc.status == PROC_CANCELLED`` (the second call short-circuits at
    the top of ``OffboardingService.cancel_process``), but the per-seat
    ``enqueue_grant_for_seat`` call inside the cancel path is NOT
    idempotent — every duplicate run would create another grant task
    for the same seat. The Idempotency-Key guard sits in front of the
    whole flow and replays the cached envelope on retry instead of
    re-executing.

    Added 2026-05-22 — pairs with suspend-sso / complete, both of
    which already had the same guard.
    """
    # Per-actor isolation as defence-in-depth: even though
    # ``require_admin()`` already gates every caller, partitioning the
    # idempotency cache per-actor prevents a second admin who happens
    # to send the same client-side ``Idempotency-Key`` from receiving
    # the cached response envelope (which embeds the leaver's
    # name/work_email + cascade details).
    #
    # Actor is folded into the ``key`` (String(128)) rather than the
    # ``scope`` (String(64)) — concatenating two UUIDs into the scope
    # would exceed the column length and 500 on every cancel. Scope
    # therefore stays consistent with the sibling endpoints
    # (``offboarding.suspend_sso:{id}``, ``offboarding.complete:{id}``).
    # Pre-launch security review finding 2026-05-22.
    namespaced_key = (
        f"{idempotency_key}:{current_user.id}"
        if idempotency_key is not None else None
    )
    idem = IdempotencyService(
        db,
        scope=f"offboarding.cancel:{process_id}",
        key=namespaced_key,
    )
    cached = await idem.lookup()
    if cached is not None:
        return cached

    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    try:
        await svc.cancel_process(
            process_id,
            reason=body.reason,
            trace_id=trace_id,
            actor_user_id=current_user.id,
            confirm_irreversible=body.confirm_irreversible,
        )
        await db.commit()
        # #1794: reverse the Google account suspend AFTER the cancel is
        # durably committed — never inside the cancel transaction (a rollback
        # there would leave the ledger 'suspended' on a re-enabled account).
        # Best-effort + self-committing; never raises.
        await svc.unsuspend_google_after_cancel(
            process_id, trace_id=trace_id, actor_user_id=current_user.id,
        )
        # Same post-commit contract for the Zoom day-of deactivate reversal —
        # a re-activated (mis-flagged) leaver must not stay locked out of Zoom.
        # Best-effort + self-committing; never raises.
        await svc.unsuspend_zoom_after_cancel(
            process_id, trace_id=trace_id, actor_user_id=current_user.id,
        )
    except Exception:
        await idem.mark_failed()
        raise
    response = success_response(
        data=await svc.get_process_detail(process_id), request=request,
    )
    await idem.record(response, resource_id=process_id)
    return response


# ─── Blockers ───


@router.post("/blockers/{blocker_id}/revoke")
async def revoke_blocker(
    request: Request,
    blocker_id: UUID,
    body: BlockerRevokeRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    await svc.resolve_blocker_by_revoke(
        blocker_id,
        notes=body.notes,
        trace_id=trace_id,
        actor_user_id=current_user.id,
        successor_user_id=body.successor_user_id,
    )
    await db.commit()
    return success_response(data=None, request=request)


@router.get("/blockers/{blocker_id}/revoke-preview")
async def revoke_blocker_preview(
    request: Request,
    blocker_id: UUID,
    current_user: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Tells the UI whether this blocker's revoke needs a successor
    picker, and suggests the leaver's manager as the default pick.
    """
    svc = OffboardingService(db)
    data = await svc.get_blocker_revoke_preview(blocker_id)
    return success_response(data=data, request=request)


@router.post("/blockers/{blocker_id}/mark-resolved")
async def mark_blocker_resolved(
    request: Request,
    blocker_id: UUID,
    body: BlockerRevokeRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    await svc.mark_blocker_resolved_manual(
        blocker_id, notes=body.notes, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=None, request=request)


@router.post("/blockers/{blocker_id}/handover")
async def handover_blocker(
    request: Request,
    blocker_id: UUID,
    body: BlockerHandoverRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Admin hands a generic-flow blocker (SSO grant / domain
    registration) to a named successor. Rewrites the resource's
    ``person_id`` and marks the blocker ``transferred``.
    """
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    await svc.resolve_blocker_by_handover_to_successor(
        blocker_id,
        successor_person_id=body.successor_person_id,
        trace_id=trace_id,
        actor_user_id=current_user.id,
        actor=current_user,
    )
    await db.commit()
    return success_response(data=None, request=request)


@router.post("/blockers/{blocker_id}/skip")
async def skip_blocker(
    request: Request,
    blocker_id: UUID,
    body: BlockerSkipRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    await svc.skip_blocker(
        blocker_id, reason=body.reason, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=None, request=request)


# ─── Transfers ───


transfer_router = APIRouter(
    prefix="/api/v1/transfers",
    tags=["transfers"],
    dependencies=[Depends(_leaver_runtime_gate)],
)


async def _commit_transfer_then_notify(
    db: AsyncSession, svc: OffboardingService, *, trace_id: str,
) -> None:
    """Commit/release transfer locks before any notification channel call."""
    await db.commit()
    await svc.dispatch_transfer_notifications_after_commit(trace_id=trace_id)


@transfer_router.post("")
async def create_transfer(
    request: Request,
    body: AssetTransferCreateRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    xfer = await svc.create_transfer(
        current_user.id,
        recipient_id=body.recipient_id,
        asset_type=body.asset_type,
        asset_ref_id=body.asset_ref_id,
        reason=body.reason,
        offboarding_process_id=body.offboarding_process_id,
        trace_id=trace_id,
    )
    await _commit_transfer_then_notify(db, svc, trace_id=trace_id)
    # Enrich
    people = await svc._load_people({xfer.initiator_id, xfer.recipient_id})
    return success_response(data=svc._to_transfer_response(xfer, people), request=request)


@transfer_router.get("/suggest-recipient")
async def suggest_transfer_recipient(
    request: Request,
    asset_type: str = Query(..., min_length=1),
    asset_ref_id: UUID = Query(...),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Return a suggested transfer recipient for an asset.

    Round-13: hardware in a configured office (asset.location matches
    asset_office_addresses.city, strict equality) suggests the local
    office admin (e.g. Sunnyvale → Kristine Ye). Cross-geo offboarding
    is the dominant real-world flow; without this hint the transfer
    modal forces the leaver to remember their local 行政's name. Other
    asset types currently have no suggestion (returns ``null``); the
    UI can fall back to a free-text user picker.

    Response shape::

        {
            "person_id": "...",
            "display_name": "Kristine Ye",
            "work_email": "...",
            "reason": "location_admin",       # why suggested
            "location_name": "美国",           # the asset's location
        }

    Or ``null`` if no suggestion. Auth: caller must currently HOLD the
    asset (assigned_to or borrowed_by) OR be an asset admin. Without
    this scope the endpoint becomes an org-chart enumerator — any
    authenticated employee could iterate asset UUIDs (~600 in deploy)
    and harvest every office admin's display_name + work_email + city.
    Pre-launch P1 fix (Round 45, CR-410).
    """
    if asset_type != "hardware":
        return success_response(data=None, request=request)
    asset = await db.get(AssetItem, asset_ref_id)
    if asset is None or not asset.location:
        return success_response(data=None, request=request)

    # Authorisation: caller must be holding the asset OR be an asset
    # admin / IT admin / management.
    from app.core.rbac import can_read_all_assets

    holds = (
        asset.assigned_to == current_user.id
        or asset.borrowed_by == current_user.id
    )
    if not holds and not can_read_all_assets(current_user.role_names):
        return success_response(data=None, request=request)
    from app.domains.asset.location_admin_sync import (
        resolve_location_primary_admins_bulk,
    )
    from app.domains.asset.location_resolver import LocationIndex

    index = await LocationIndex.load(db)
    location_id = index.resolve_id(asset.location)
    if location_id is None:
        return success_response(data=None, request=request)
    primaries = await resolve_location_primary_admins_bulk(db, [location_id])
    admin_id = primaries.get(location_id)
    if admin_id is None:
        return success_response(data=None, request=request)
    from app.identity.person_models import IdentityPerson

    admin = await db.get(IdentityPerson, admin_id)
    if (
        admin is None
        or admin.deleted_at is not None
        or admin.employment_status != "active"
    ):
        return success_response(data=None, request=request)
    location_name = index.resolve_name(asset.location)
    return success_response(
        data={
            "person_id": str(admin.id),
            "display_name": admin.display_name or admin.work_email,
            "work_email": admin.work_email,
            # The suggestion follows the asset's LOCATION now, so the
            # response says so. Nothing consumes this string today —
            # the transfer modal renders the name — and leaving it as
            # "office_admin" would have been a label that contradicts
            # how the person was chosen.
            "reason": "location_admin",
            "location_name": location_name,
        },
        request=request,
    )


@transfer_router.get("/incoming")
async def list_incoming(
    request: Request,
    status: str | None = Query(None),
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    items = await svc.list_transfers_for_user(
        current_user.id, direction="incoming", status=status,
        limit=page_size, offset=(page - 1) * page_size,
    )
    total = await svc.count_transfers_for_user(
        current_user.id, direction="incoming", status=status,
    )
    return success_response(
        data=items,
        meta={"total": total, "page": page, "page_size": page_size},
        request=request,
    )


@transfer_router.get("/outgoing")
async def list_outgoing(
    request: Request,
    status: str | None = Query(None),
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    items = await svc.list_transfers_for_user(
        current_user.id, direction="outgoing", status=status,
        limit=page_size, offset=(page - 1) * page_size,
    )
    total = await svc.count_transfers_for_user(
        current_user.id, direction="outgoing", status=status,
    )
    return success_response(
        data=items,
        meta={"total": total, "page": page, "page_size": page_size},
        request=request,
    )


@transfer_router.post("/{transfer_id}/accept")
async def accept_transfer(
    request: Request,
    transfer_id: UUID,
    body: AssetTransferResponseBody,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    xfer = await svc.respond_to_transfer(
        transfer_id,
        accept=True,
        response_note=body.response_note,
        acting_user_id=current_user.id,
        trace_id=trace_id,
    )
    await _commit_transfer_then_notify(db, svc, trace_id=trace_id)
    people = await svc._load_people({xfer.initiator_id, xfer.recipient_id})
    return success_response(data=svc._to_transfer_response(xfer, people), request=request)


@transfer_router.post("/{transfer_id}/reject")
async def reject_transfer(
    request: Request,
    transfer_id: UUID,
    body: AssetTransferResponseBody,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    xfer = await svc.respond_to_transfer(
        transfer_id,
        accept=False,
        response_note=body.response_note,
        acting_user_id=current_user.id,
        trace_id=trace_id,
    )
    await _commit_transfer_then_notify(db, svc, trace_id=trace_id)
    people = await svc._load_people({xfer.initiator_id, xfer.recipient_id})
    return success_response(data=svc._to_transfer_response(xfer, people), request=request)


@transfer_router.post("/{transfer_id}/cancel")
async def cancel_transfer(
    request: Request,
    transfer_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    xfer = await svc.cancel_transfer(
        transfer_id, acting_user_id=current_user.id, trace_id=trace_id,
    )
    await db.commit()
    people = await svc._load_people({xfer.initiator_id, xfer.recipient_id})
    return success_response(data=svc._to_transfer_response(xfer, people), request=request)


# Pre-launch P2 fix (Round 1, SEC-007): admin-override cancel for
# orphaned peer-to-peer transfers. After the initiator becomes a
# leaver, ``offboarding_process_id IS NULL`` transfers had no escape
# path — recipient saw stale rows in /my/transfers forever.
@transfer_router.post("/{transfer_id}/admin-cancel")
async def admin_cancel_transfer(
    request: Request,
    transfer_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = OffboardingService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    xfer = await svc.cancel_transfer(
        transfer_id,
        acting_user_id=current_user.id,
        trace_id=trace_id,
        is_admin_override=True,
    )
    await db.commit()
    people = await svc._load_people({xfer.initiator_id, xfer.recipient_id})
    return success_response(data=svc._to_transfer_response(xfer, people), request=request)


# ─── Account end-state assurance ───
#
# The question these three endpoints answer is not "did Helix revoke
# things" (the rest of this router covers that) but "can this leaver
# still sign in". They read a table refreshed by the per-system leaver
# sweeps rather than calling vendors inline, so the page loads in one
# query and cannot be made slow — or rate-limited — by how many leavers
# the company has.


@router.get("/account-assurance/summary")
async def account_assurance_summary(
    request: Request,
    _reader: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """KPI row + identity-chain health for the 账号终态 page header.

    Includes ``evidence_age_hours`` on purpose: a governance page whose
    numbers are three days stale is more dangerous than one with no
    numbers, because it looks like an answer. The header always says how
    old the evidence is.
    """
    from app.domains.offboarding import assurance_service

    return success_response(
        data=await assurance_service.summary(db), request=request,
    )


@router.get("/account-assurance")
async def list_account_assurance(
    request: Request,
    verdict: str | None = Query(None, pattern="^(ok|at_risk|exempt|unknown)$"),
    # ``isync:<slug>`` joined the vocabulary in PR3 (one system per
    # revocation-enabled identity-sync target), so the filter can no longer
    # be a two-value enum. Bounded by length + charset instead; an unknown
    # value simply matches nothing.
    system: str | None = Query(None, max_length=32, pattern=r"^[a-z0-9_:.-]+$"),
    q: str | None = Query(None, max_length=120),
    _reader: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Leavers with their per-system end state, worst first.

    Grouped by person rather than by row: "this person's Google AND
    Atlassian are both still open" is one problem somebody owns
    end-to-end, and splitting it across two table rows makes the reader
    do the join.
    """
    from app.domains.offboarding import assurance_service

    items = await assurance_service.list_people(
        db, verdict=verdict, system=system, q=q,
    )
    return success_response(data=items, request=request)


@router.post("/account-assurance/{person_id}/enforce")
async def enforce_account_assurance(
    request: Request,
    person_id: UUID,
    systems: str = Query(
        "google,atlassian",
        max_length=512,
        description=(
            "Comma-separated systems to close: google, atlassian, and/or "
            "isync:<target slug> for an identity-sync target (PR3)."
        ),
    ),
    override_exemption: bool = Query(
        False,
        description=(
            "Proceed even though a retention decision covers this account. "
            "Separate flag so reversing somebody's decision can never "
            "happen by accident — the UI asks for it explicitly."
        ),
    ),
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Close this leaver's accounts now, then read the result back.

    Admin-only and audited. Refuses on anyone whose employment status is
    not ``left`` — the single unrecoverable outcome here is locking out
    somebody who still works here, and that check reads live person
    state rather than the (possibly stale) assurance row.

    Refuses on an account carrying a retention decision: reversing
    somebody's deliberate "keep this open" is a people decision, not a
    platform one. Clear the registry row first if the decision no longer
    holds.
    """
    from fastapi import HTTPException

    from app.domains.offboarding import assurance_service

    wanted = [
        s.strip() for s in (systems or "").split(",") if s.strip()
    ]
    if not wanted:
        raise HTTPException(status_code=400, detail="no systems selected")
    trace_id = (
        str(request.state.trace_id)
        if hasattr(request.state, "trace_id") else ""
    )
    try:
        result = await assurance_service.enforce_person(
            db, person_id=person_id, systems=wanted,
            actor_user_id=current_user.id, trace_id=trace_id,
            override_exemption=override_exemption,
        )
    except assurance_service.AssuranceTargetNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except assurance_service.AssuranceEnforceError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    await db.commit()
    return success_response(data=result, request=request)