"""HTTP surface for the Leaver Blocked dashboard + dual-sign override.

All routes mount under ``/api/v1/leaver-blocked``. RBAC:

  * ``can_manage_leaver_blocked`` (global_admin / hr) — initiate
    overrides, view full audit log, force-advance leavers.
  * ``can_secondary_sign_leaver_override`` — secondary signer
    approve/reject (additionally validated by manager-chain check
    inside the service).
  * Read endpoints (dashboard summary / list / detail) — same
    ``can_manage_leaver_blocked`` gate.
"""
from __future__ import annotations

import uuid
from typing import Any
from uuid import UUID

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

from app.audit.service import AuditService
from app.core.database import get_db
from app.core.dependencies import get_current_user, get_real_ip
from app.core.exceptions import ForbiddenError, NotFoundError
from app.core.idempotency import IdempotencyService
from app.core.rbac import (
    can_force_bypass_leaver_grace,
    can_manage_leaver_blocked,
)
from app.core.response import success_response
from app.domains.leaver_blocked.dashboard_service import (
    LeaverBlockedDashboardService,
)
from app.domains.leaver_blocked.models import (
    LeaverBlockAudit,
    LeaverDualSignOverride,
)
from app.domains.leaver_blocked.override_service import (
    DualSignOverrideService,
)
from app.domains.leaver_blocked.schemas import (
    CancelOverrideIn,
    ForceBypassGraceIn,
    InitiateOverrideIn,
    LeaverBlockedDetailOut,
    LeaverBlockedListOut,
    LeaverBlockedRowOut,
    LeaverBlockedSummaryOut,
    SecondaryDecisionIn,
)
from app.identity.person_models import IdentityPerson

router = APIRouter(
    prefix="/api/v1/leaver-blocked",
    tags=["leaver-blocked"],
)


def _require_manage(user) -> None:
    if not can_manage_leaver_blocked(user.role_names):
        raise ForbiddenError(
            "global_admin or hr role required to access leaver-blocked dashboard"
        )


# ── Dashboard ───────────────────────────────────────────────────────


@router.get("/summary")
async def get_summary(
    request: Request,
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_manage(user)
    summary = await LeaverBlockedDashboardService(db).get_summary()
    payload = LeaverBlockedSummaryOut(**summary.__dict__)
    return success_response(data=payload.model_dump(), request=request)


@router.get("/list")
async def list_blocked(
    request: Request,
    sla_level: str | None = Query(default=None),
    keyword: str | None = Query(default=None),
    limit: int = Query(default=50, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_manage(user)
    rows, total = await LeaverBlockedDashboardService(db).list_blocked(
        sla_level=sla_level, keyword=keyword,
        limit=limit, offset=offset,
    )
    items = [LeaverBlockedRowOut(**r.__dict__) for r in rows]
    payload = LeaverBlockedListOut(items=items, total=total)
    return success_response(
        data=payload.model_dump(),
        meta={"total": total, "limit": limit, "offset": offset},
        request=request,
    )


@router.get("/processes/{process_id}/detail")
async def get_detail(
    process_id: UUID,
    request: Request,
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_manage(user)
    try:
        detail = await LeaverBlockedDashboardService(db).get_detail(process_id)
    except ValueError as exc:
        raise NotFoundError("offboarding_process", str(process_id)) from exc
    payload = LeaverBlockedDetailOut(**detail)
    return success_response(data=payload.model_dump(), request=request)


# ── Audit log ──────────────────────────────────────────────────────


@router.get("/audit")
async def list_audit(
    request: Request,
    person_id: UUID | None = Query(default=None),
    denied_reason: str | None = Query(default=None),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_manage(user)
    conds = []
    if person_id is not None:
        conds.append(LeaverBlockAudit.person_id == person_id)
    if denied_reason:
        conds.append(LeaverBlockAudit.denied_reason == denied_reason)
    # Full filtered count so the client can paginate (the table grows
    # unbounded — one row per blocked attempt — and offset/limit alone
    # gave no way to know whether more pages exist).
    total = (await db.execute(
        select(func.count()).select_from(LeaverBlockAudit).where(*conds),
    )).scalar_one()
    query = (
        select(LeaverBlockAudit).where(*conds)
        .order_by(LeaverBlockAudit.attempted_at.desc())
        .offset(offset).limit(limit)
    )
    rows = list((await db.execute(query)).scalars().all())
    return success_response(
        data=[
            {
                "id": str(r.id),
                "person_id": str(r.person_id),
                "attempted_at": r.attempted_at.isoformat(),
                "attempted_by_user_id": (
                    str(r.attempted_by_user_id) if r.attempted_by_user_id else None
                ),
                "attempted_via": r.attempted_via,
                "target_status": r.target_status,
                "denied_reason": r.denied_reason,
                "pending_blocker_count": r.pending_blocker_count,
                "pending_blockers": r.pending_blockers,
                "detail": r.detail,
                "client_ip": r.client_ip,
            }
            for r in rows
        ],
        meta={"limit": limit, "offset": offset, "total": total},
        request=request,
    )


# ── Dual-sign override ─────────────────────────────────────────────


@router.post("/overrides")
async def initiate_override(
    body: InitiateOverrideIn,
    request: Request,
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    _require_manage(user)
    trace_id = (
        getattr(request.state, "trace_id", None) or uuid.uuid4()
    )
    if isinstance(trace_id, str):
        trace_id = uuid.UUID(trace_id)
    service = DualSignOverrideService(db)
    override = await service.initiate(
        person_id=body.person_id,
        blocker_ids=body.blocker_ids,
        primary_reason=body.primary_reason,
        secondary_signer_user_id=body.secondary_signer_user_id,
        primary_signer=user,
        trace_id=trace_id,
        client_ip=get_real_ip(request),
    )
    return success_response(
        data=await _serialize_override(db, override),
        request=request,
    )


@router.post("/overrides/{override_id}/approve")
async def approve_override(
    override_id: UUID,
    body: SecondaryDecisionIn,
    request: Request,
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Authorisation: the service-layer ``secondary_approve`` call
    # enforces ``override.secondary_signer_user_id == user.id`` and
    # raises ForbiddenError on mismatch. The role gate
    # ``can_secondary_sign_leaver_override`` (global_admin / hr) is
    # the BASELINE — anyone outside it must be the named secondary
    # signer or the call rejects. We let the request through here
    # so the service can do the precise identity match against the
    # row; pre-rejecting on role would lock out legitimate
    # named-secondary signers (who may be a department director
    # without admin / hr role).
    trace_id = getattr(request.state, "trace_id", None) or uuid.uuid4()
    if isinstance(trace_id, str):
        trace_id = uuid.UUID(trace_id)
    service = DualSignOverrideService(db)
    override = await service.secondary_approve(
        override_id=override_id,
        secondary_reason=body.secondary_reason,
        secondary_signer=user,
        trace_id=trace_id,
        client_ip=get_real_ip(request),
    )
    return success_response(
        data=await _serialize_override(db, override),
        request=request,
    )


@router.post("/overrides/{override_id}/reject")
async def reject_override(
    override_id: UUID,
    body: SecondaryDecisionIn,
    request: Request,
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    trace_id = getattr(request.state, "trace_id", None) or uuid.uuid4()
    if isinstance(trace_id, str):
        trace_id = uuid.UUID(trace_id)
    service = DualSignOverrideService(db)
    override = await service.secondary_reject(
        override_id=override_id,
        secondary_reason=body.secondary_reason,
        secondary_signer=user,
        trace_id=trace_id,
        client_ip=get_real_ip(request),
    )
    return success_response(
        data=await _serialize_override(db, override),
        request=request,
    )


@router.post("/overrides/{override_id}/cancel")
async def cancel_override(
    override_id: UUID,
    body: CancelOverrideIn,
    request: Request,
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Authorisation: the service-layer ``cancel`` call enforces that
    # ``actor.id == override.primary_signer_user_id`` OR the actor has
    # ``can_manage_leaver_blocked`` (global_admin / hr).  We add a
    # router-level baseline here: an employee with neither the role
    # nor the primary-signer identity cannot cancel overrides.  Loading
    # the override to check primary_signer_user_id before the service
    # call would add an extra DB round-trip; instead we rely on the
    # service as the canonical identity gate and add a role-baseline
    # pre-check that passes if the caller has manage rights (fast path)
    # OR unconditionally defers to the service (which will 403 if the
    # caller is an unrelated employee).
    # The design mirrors ``approve_override`` / ``reject_override`` above
    # (comments at line ~210) — the named primary signer is the only
    # non-admin allowed, and the service enforces that.
    trace_id = getattr(request.state, "trace_id", None) or uuid.uuid4()
    if isinstance(trace_id, str):
        trace_id = uuid.UUID(trace_id)
    service = DualSignOverrideService(db)
    override = await service.cancel(
        override_id=override_id,
        actor=user,
        reason=body.reason,
        trace_id=trace_id,
        client_ip=get_real_ip(request),
    )
    return success_response(
        data=await _serialize_override(db, override),
        request=request,
    )


@router.post("/overrides/{override_id}/execute")
async def execute_override(
    override_id: UUID,
    request: Request,
    idempotency_key: str | None = Header(
        None, alias="Idempotency-Key", max_length=128),
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Execute an already-approved override.

    Idempotency: scoped per ``override_id`` so a double-click cannot
    fire the override twice. Replay returns the cached response. The
    server-side state machine (``status == approved``) is the
    last-line defence — idempotency is the *first* line so a
    duplicate POST never reaches the state-flip code.
    """
    _require_manage(user)
    idem = IdempotencyService(
        db,
        scope=f"leaver_blocked.execute_override:{override_id}",
        key=idempotency_key,
    )
    cached = await idem.lookup()
    if cached is not None:
        return cached

    trace_id = getattr(request.state, "trace_id", None) or uuid.uuid4()
    if isinstance(trace_id, str):
        trace_id = uuid.UUID(trace_id)
    service = DualSignOverrideService(db)
    try:
        override = await service.execute(
            override_id=override_id,
            actor=user,
            trace_id=trace_id,
            client_ip=get_real_ip(request),
            user_agent=request.headers.get("User-Agent"),
        )
    except Exception:
        await idem.mark_failed()
        raise

    response = success_response(
        data=await _serialize_override(db, override),
        request=request,
    )
    await idem.record(response, resource_id=override_id)
    return response


@router.get("/overrides")
async def list_overrides(
    request: Request,
    status_filter: str | None = Query(default=None, alias="status"),
    person_id: UUID | None = Query(default=None),
    only_my_secondary: bool = Query(default=False),
    limit: int = Query(default=100, ge=1, le=500),
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    statuses = [s.strip() for s in status_filter.split(",")] if status_filter else None
    secondary_filter = user.id if only_my_secondary else None

    # Non-admin callers can only see overrides where they are the
    # named secondary signer, so they can act on their queue.
    if not can_manage_leaver_blocked(user.role_names):
        secondary_filter = user.id

    service = DualSignOverrideService(db)
    rows = await service.list_overrides(
        statuses=statuses,
        person_id=person_id,
        secondary_signer_user_id=secondary_filter,
        limit=limit,
    )
    total = await service.count_overrides(
        statuses=statuses,
        person_id=person_id,
        secondary_signer_user_id=secondary_filter,
    )
    return success_response(
        data=[await _serialize_override(db, o) for o in rows],
        meta={"total": total, "limit": limit},
        request=request,
    )


@router.get("/overrides/{override_id}")
async def get_override(
    override_id: UUID,
    request: Request,
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    override = await db.get(LeaverDualSignOverride, override_id)
    if override is None:
        raise NotFoundError("leaver_dual_sign_override", str(override_id))
    if not can_manage_leaver_blocked(user.role_names):
        # Non-admin: only allow if the user is the named secondary
        if override.secondary_signer_user_id != user.id:
            raise ForbiddenError(
                "Only global_admin/hr or the named secondary signer "
                "can read this override"
            )
    return success_response(
        data=await _serialize_override(db, override),
        request=request,
    )


# ── force-advance (run the tick on a single person now) ────────────


async def _run_advance_for_process(
    *,
    process_id: UUID,
    request: Request,
    idempotency_key: str | None,
    user: IdentityPerson,
    db: AsyncSession,
    bypass_grace: bool,
    audit_action: str,
    extra_audit_detail: dict[str, Any] | None = None,
) -> Any:
    """Shared advance pipeline used by ``advance_now`` and
    ``force_bypass_grace_advance``.

    Differences between the two callers live entirely in the
    arguments:

      * ``bypass_grace`` — plumbed into ``_advance_single``; True
        collapses the 24h IT grace window.
      * ``audit_action`` — distinct action string per endpoint so
        forensics can tell "regular re-run" from "bypass exercised".
      * ``extra_audit_detail`` — bypass path attaches the required
        ``reason`` here; advance_now passes None.

    Everything else (idempotency scope keyed by audit_action,
    process/person load, savepoint, summary buckets, durable failure
    audit, identity-sync hook, success audit) is identical.
    """
    idem = IdempotencyService(
        db,
        scope=f"{audit_action}:{process_id}",
        key=idempotency_key,
    )
    cached = await idem.lookup()
    if cached is not None:
        return cached

    from app.domains.offboarding.models import OffboardingProcess
    proc = await db.get(OffboardingProcess, process_id)
    if proc is None:
        await idem.mark_failed()
        raise NotFoundError("offboarding_process", str(process_id))
    person = await db.get(IdentityPerson, proc.person_id)
    if person is None:
        await idem.mark_failed()
        raise NotFoundError("identity_person", str(proc.person_id))

    # Run the standard advance pipeline scoped to THIS person.
    from app.hr_lifecycle.advance_service import ProcessAdvanceService

    # Round-21 (2026-05-13) prod incident: this was importing
    # ``get_platform_setting`` from ``app.core.config`` where the
    # symbol does not exist (canonical location is
    # ``app.integrations.service`` — see advance_service.py line 106
    # for the same import used everywhere else). The ImportError
    # raised on every "立即推进" click BEFORE the try/except below
    # could run, so the failure bubbled straight to FastAPI's
    # generic ``_unhandled_exception_handler`` as the user-visible
    # "An unexpected error occurred." 500 with no audit trail.
    from app.integrations.service import get_platform_setting
    service = ProcessAdvanceService(db)
    cfg_fallback = await get_platform_setting(
        "lifecycle.fallback_owner_email", default=None, session=db,
    )
    resolved_fallback = (cfg_fallback or "").strip() or None

    transitioned_now = False
    summary: dict[str, object] = {
        "processed": 1,
        "force_transferred_blockers": 0,
        "transitioned_to_left": 0,
        "grace_held": 0,
        "held_missing_fallback": 0,
        "held_pending_blockers": 0,
        # Round-58 (2026-05-30): software_* blocker waiting on its
        # ProvisioningTask callback. Distinct from ``held_missing_fallback``
        # so the admin "立即推进" button can show "system is working,
        # try again after the task fires" instead of the misleading
        # "no fallback configured" Round-21 was meant to eliminate.
        "held_awaiting_provisioning": 0,
        "pending_blocker_count": 0,
        "incidents_created": 0,
        "errors": [],
        # 2026-06-02: echo the bypass flag so post-hoc forensics on
        # this summary alone can tell which code path produced it
        # (the audit row's ``action`` is the canonical signal, but
        # the summary is what gets returned to the FE toast — having
        # the flag locally avoids a second audit_logs round-trip).
        "bypass_grace": bypass_grace,
    }
    try:
        async with db.begin_nested():
            result = await service._advance_single(
                person,
                fallback_email=resolved_fallback,
                bypass_grace=bypass_grace,
            )
        summary["force_transferred_blockers"] = result.get(
            "force_transferred", 0,
        )
        if result.get("transitioned"):
            summary["transitioned_to_left"] = 1
            transitioned_now = True
        elif result.get("grace_held"):
            summary["grace_held"] = 1
        else:
            # Round-21 (2026-05-13): split the previous catch-all
            # ``held_missing_fallback`` flag into two distinct outcomes.
            # The old shape mislabelled "force-transfer rejected every
            # blocker so we couldn't transition" (the huohao@ /
            # frank.shen@ prod incident — 14 cloud_access_grant blockers
            # held by RBAC) as "no fallback configured", which sent
            # admins on a wild goose chase configuring fallback emails
            # that were already set. ``hold_reason`` from
            # ``_advance_single`` is the authoritative discriminator.
            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): software_* row waiting on its
                # ProvisioningTask not-yet-fired. NOT operator-actionable
                # — the next advance tick after the task completes will
                # transition cleanly. Distinct from missing_fallback so
                # the admin button surfaces the right status to UI.
                summary["held_awaiting_provisioning"] = 1
                summary["pending_blocker_count"] = result.get(
                    "pending_blocker_count", 0,
                )
            else:
                # missing_fallback or unknown — preserve old semantics so
                # any callers grepping the field name still find it.
                summary["held_missing_fallback"] = 1
        summary["incidents_created"] = result.get("incidents_created", 0)
    except Exception as exc:  # noqa: BLE001
        # Round-20 post-fix follow-up: surface the failure as a real
        # HTTP error so monitoring + the admin button see status !=
        # 2xx, and write the audit row with status="failure" not
        # "success". The old shape's silent-success was the very
        # property the round-20 advance_now fix was trying to remove.
        #
        # Round-21 (2026-05-13) prod incident: the failure audit row
        # below used to be added to the request-scoped session and was
        # silently rolled back by ``get_db``'s ``session.rollback()``
        # on the bubbling exception — leaving zero forensic trail of
        # admin "立即推进" failures in audit_logs (the huohao@ / frank.
        # shen@ incident left audit completely empty, sending us to
        # blocker.last_error_message instead). Use an independent
        # session that commits in its own transaction so the row
        # survives the outer rollback — mirrors the durable-audit
        # pattern in ``cmdb/lifecycle_service.py`` (~line 2195) for
        # ``cloud.grant.revoke_attempt_failed``. Best-effort: if the
        # durable audit itself fails we log and continue to re-raise
        # the original exception so the user/monitoring still sees the
        # 5xx (masking the real error with the audit-failure error
        # would be the worst possible outcome).
        # This handler ends in an unconditional ``raise`` and the durable
        # audit below runs on its OWN session, so the caller's session is
        # never used again after this line. It is also very likely already
        # poisoned by the exception being handled — exactly when opening a
        # savepoint would raise and mask the real error.
        # savepoint-exempt: handler re-raises; durable audit uses its own session
        try:
            await idem.mark_failed()
        except Exception:  # noqa: BLE001
            pass
        trace_id_str = (
            getattr(request.state, "trace_id", None)
            or str(uuid.uuid4())
        )
        try:
            from app.core.database import async_session_factory  # noqa: PLC0415
            from app.core.exceptions import format_exception_for_log  # noqa: PLC0415
            failure_detail: dict[str, Any] = {
                "person_id": str(person.id),
                "error_type": type(exc).__name__,
                "error_code": getattr(exc, "code", None),
                # ``format_exception_for_log`` captures
                # AppException.message (where the user-facing
                # text actually lives) — a bare ``str(exc)`` is
                # empty for AppException because the class
                # deliberately skips ``Exception.__init__(args)``.
                "error": format_exception_for_log(
                    exc, max_len=500, include_type=False,
                ),
            }
            if extra_audit_detail:
                failure_detail.update(extra_audit_detail)
            async with async_session_factory() as audit_db:
                await AuditService(audit_db).log(
                    trace_id=trace_id_str,
                    actor_type="user",
                    actor_user_id=user.id,
                    action=audit_action,
                    resource_type="offboarding_process",
                    resource_id=str(process_id),
                    status="failure",
                    detail=failure_detail,
                )
                await audit_db.commit()
        except Exception:  # noqa: BLE001
            from app.core.safe_logging import get_safe_logger  # noqa: PLC0415
            get_safe_logger(__name__).exception(
                "durable %s failure audit write failed "
                "(process_id=%s, original error suppressed in audit "
                "but still raised to caller)",
                audit_action, process_id,
            )
        raise

    # Mirror the post-commit identity-sync event-hook that ``run_tick``
    # fires for newly-transitioned persons. The hook opens a fresh
    # session so we must commit the outer session FIRST.
    if transitioned_now:
        await db.commit()
        from app.core.background import spawn
        from app.core.coordinators.leaver_cloud_sweep_coordinator import (
            sweep_then_offboard_sync,
        )
        # ``spawn`` retains a strong reference until completion so the
        # post-commit work can't be GC'd mid-flight. Sweep the leaver's
        # residual AWS Identity Center assignments BEFORE the offboard
        # directory sync's SCIM deprovision deletes their Identity Store
        # user — otherwise those untracked assignments become blank-name
        # orphans (mirrors run_tick's post-commit fan-out).
        spawn(
            sweep_then_offboard_sync(person.id),
            name=f"offboard_sweep_sync:{person.id}",
        )

    success_detail: dict[str, Any] = {
        "person_id": str(person.id), "summary": summary,
    }
    if extra_audit_detail:
        success_detail.update(extra_audit_detail)
    audit = AuditService(db)
    await audit.log(
        trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()),
        actor_type="user",
        actor_user_id=user.id,
        action=audit_action,
        resource_type="offboarding_process",
        resource_id=str(process_id),
        status="success",
        detail=success_detail,
    )

    # Round-62 P1 fix: include the top-level ``transitioned`` boolean
    # so the cockpit "立即推进" toast renders "推进:是/否" accurately.
    # Without this, ``res.transitioned`` was undefined in the FE
    # wrapper and the toast always showed "推进:否" regardless of
    # whether the person actually transitioned to status='left'.
    response = success_response(
        data={"summary": summary, "transitioned": transitioned_now},
        request=request,
    )
    await idem.record(response, resource_id=process_id)
    return response


@router.post("/processes/{process_id}/advance-now")
async def advance_now(
    process_id: UUID,
    request: Request,
    idempotency_key: str | None = Header(
        None, alias="Idempotency-Key", max_length=128),
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Re-run ProcessAdvanceService for one person on demand instead
    of waiting for the next 15-min tick. Lets the admin verify a
    just-completed transfer immediately surfaces.

    Idempotency: scoped per ``process_id``. A duplicate click while
    the underlying tick is still flushing returns the cached summary
    instead of running the full pipeline twice. The
    leaver-block audit append on each retry is the cost we want to
    avoid; ``IdempotencyService`` handles the cache.

    Round-20 audit (A3F3) fix: the previous shape called
    ``run_tick(today=None)`` which iterates EVERY ``leaver_pending``
    person whose ``termination_date <= today``. The audit row claimed
    the action targeted ``process_id`` while side-effects fanned out
    to unrelated leavers — wrong-target force-transfer, wrong-target
    suspend_sso, audit trail unreliable. Now scoped to the single
    process: we call ``_advance_single(person)`` directly inside its
    own savepoint so the tick semantics (atomic, per-person) are
    preserved without touching any other leaver.

    This path RESPECTS the T+0 24h IT grace window. To bypass it see
    ``POST /processes/{id}/force-bypass-grace``.
    """
    _require_manage(user)
    return await _run_advance_for_process(
        process_id=process_id,
        request=request,
        idempotency_key=idempotency_key,
        user=user,
        db=db,
        bypass_grace=False,
        audit_action="leaver_blocked.advance_now",
    )


@router.post("/processes/{process_id}/force-bypass-grace")
async def force_bypass_grace_advance(
    process_id: UUID,
    body: ForceBypassGraceIn,
    request: Request,
    idempotency_key: str | None = Header(
        None, alias="Idempotency-Key", max_length=128),
    user: IdentityPerson = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Force-advance one leaver, collapsing the T+0 24h IT grace to
    zero so physical force-transfer + revoke fire on this tick.

    Stricter RBAC than ``advance_now``: global_admin / it_admin only
    (HR cannot bypass the IT grace — the window exists for IT).

    Required ``reason`` ≥ MIN_REASON_CHARS so the audit trail records
    *why* the bypass was warranted. The canonical case is HRIS-late-
    detection: termination weeks ago, no T-N reminders ever fired,
    IT has already drained the leaver's resources manually, and IT
    wants to stop waiting another 24h before the platform finalises.

    Idempotency: distinct scope from ``advance_now`` so a click on
    each button within the same ``process_id`` does not collide.
    Audit action: ``leaver_blocked.force_bypass_grace_advance``
    (separate from ``leaver_blocked.advance_now`` so forensics can
    distinguish "regular re-run" from "bypass exercised").
    """
    if not can_force_bypass_leaver_grace(user.role_names):
        raise ForbiddenError(
            "Only IT admin (global_admin) can force-bypass the T+0 "
            "24h grace window. HR / leaver-manager roles are denied "
            "by design — the grace window exists for IT, so only IT "
            "can legitimately close it early.",
        )
    return await _run_advance_for_process(
        process_id=process_id,
        request=request,
        idempotency_key=idempotency_key,
        user=user,
        db=db,
        bypass_grace=True,
        audit_action="leaver_blocked.force_bypass_grace_advance",
        extra_audit_detail={"reason": body.reason},
    )


# ── helpers ────────────────────────────────────────────────────────


async def _serialize_override(
    db: AsyncSession, o: LeaverDualSignOverride,
) -> dict[str, Any]:
    # Resolve the (≤3 distinct) IdentityPerson display names so the
    # secondary-signer review surface shows human names instead of raw
    # UUIDs — the reviewer is typically a department director arriving
    # from a Feishu deep-link who must see WHO is leaving and WHO signed.
    # All three id columns reference IdentityPerson (see override_service
    # ``db.get(IdentityPerson, primary_signer_user_id)``).
    ids = {str(o.person_id), str(o.primary_signer_user_id)}
    if o.secondary_signer_user_id:
        ids.add(str(o.secondary_signer_user_id))
    name_rows = (await db.execute(
        select(IdentityPerson.id, IdentityPerson.display_name)
        .where(IdentityPerson.id.in_(ids))
    )).all()
    names = {str(r[0]): r[1] for r in name_rows}

    def _disp(pid: Any) -> str | None:
        return names.get(str(pid)) if pid is not None else None

    return {
        "id": str(o.id),
        "process_id": str(o.process_id),
        "person_id": str(o.person_id),
        "person_display": _disp(o.person_id),
        "status": o.status,
        "primary_signer_user_id": str(o.primary_signer_user_id),
        "primary_signer_display": _disp(o.primary_signer_user_id),
        "secondary_signer_display": _disp(o.secondary_signer_user_id),
        "primary_signed_at": (
            o.primary_signed_at.isoformat() if o.primary_signed_at else None
        ),
        "primary_reason": o.primary_reason,
        "secondary_signer_user_id": (
            str(o.secondary_signer_user_id)
            if o.secondary_signer_user_id else None
        ),
        "secondary_signed_at": (
            o.secondary_signed_at.isoformat()
            if o.secondary_signed_at else None
        ),
        "secondary_reason": o.secondary_reason,
        "overridden_blocker_ids": list(o.overridden_blocker_ids or []),
        "overridden_blockers_snapshot": list(o.overridden_blockers_snapshot or []),
        "executed_at": o.executed_at.isoformat() if o.executed_at else None,
        "rejected_at": o.rejected_at.isoformat() if o.rejected_at else None,
        "rejected_reason": o.rejected_reason,
        "expires_at": o.expires_at.isoformat() if o.expires_at else None,
        "created_at": o.created_at.isoformat() if o.created_at else None,
        "updated_at": o.updated_at.isoformat() if o.updated_at else None,
    }