import hashlib
import hmac
import json
from datetime import UTC, date, datetime, timedelta
from typing import NamedTuple
from uuid import UUID
from zoneinfo import ZoneInfo

from fastapi import APIRouter, Depends, Header, Query, Request
from pydantic import BaseModel, Field, ValidationError
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.background import spawn as _spawn_background
from app.core.database import get_db
from app.core.dependencies import (
    check_login_rate_limit,
    get_current_user,
    get_real_ip,
    require_active_employee,
    require_hr_admin,
    require_hr_read,
)
from app.core.exceptions import (
    AppException,
    ForbiddenError,
    NotFoundError,
    UnauthorizedError,
)
from app.core.exceptions import (
    ValidationError as AppValidationError,
)
from app.core.person_scope import SubordinateScope, subordinate_person_ids
from app.core.rbac import GLOBAL_ADMIN_ROLE, HR_ROLE, can_view_hr_data, has_role
from app.core.response import success_response
from app.core.safe_logging import get_safe_logger
from app.core.txn import best_effort
from app.domains.hr.leave_insights_models import (  # noqa: F401 — register on Base.metadata
    HrLeaveInsightSnapshot,
)
from app.domains.hr.leave_insights_schemas import (
    CompanyBaseline,
    GroupInsightRow,
    GroupMembersResponse,
    MemberRow,
    OverviewResponse,
    TrendPoint,
    TrendResponse,
)
from app.domains.hr.leave_insights_service import (
    COMPANY_KEY,
    DIMENSION_BU,
    DIMENSION_COMPANY,
    DIMENSION_DEPARTMENT,
    DIMENSION_LEAD,
    HEALTH_ATTENTION,
    HEALTH_HEALTHY,
    HEALTH_INSUFFICIENT,
    HEALTH_NORMAL,
    HEALTH_UNCOVERED,
    LeaveInsightsService,
)
from app.domains.hr.leave_service import HrLeaveService
from app.domains.hr.schemas import (
    IncrementalSyncRequest,
    LeaveAccountResponse,
    LeavePostingResponse,
    LifecycleOverviewResponse,
    ManualMarkLeaverRequest,
    ManualMarkLeaverResponse,
    OffboardingAssetResponse,
    OffboardingAssetStatusUpdateRequest,
    OffboardingCaseResponse,
    OffboardingCompletionRequest,
    OffboardingSoftwareResponse,
    StagedEventResponse,
    SuccessFactorsWebhookRequest,
    SyncBatchResponse,
    SyncBatchSummaryResponse,
    SyncConfirmResponse,
    SyncPreviewRequest,
)
from app.domains.hr.service import (
    SuccessFactorsLifecycleService,
    run_batch_apply_in_background,
    run_leave_sync_in_background,
    run_webhook_events_in_background,
    sync_one_person_leave_now,
)
from app.identity.successfactors_client import SuccessFactorsHTTPError

logger = get_safe_logger(__name__)

router = APIRouter(prefix="/api/v1/hr", tags=["hr"])


async def _resolve_services(
    db: AsyncSession,
    *,
    hris_instance_id: UUID | None,
) -> list[SuccessFactorsLifecycleService]:
    """Resolve one or many SF services for the per-instance sync fan-out.

    With ``hris_instance_id=None`` the caller wants to sync every
    enabled instance (the "daily fan-out" semantic); with a specific
    id they want just that region. A 404 is raised when a named
    instance doesn't exist so the admin sees a clear error rather
    than a silent no-op.
    """
    from app.domains.cmdb.credential_crypto import decrypt_fields
    from app.integrations.models import Integration

    if hris_instance_id is None:
        return await SuccessFactorsLifecycleService.fan_out(db)

    row = await db.scalar(
        select(Integration).where(
            Integration.id == hris_instance_id,
            Integration.kind == "hris",
        )
    )
    if row is None or not row.enabled:
        raise AppException(
            code="INTEGRATION_NOT_FOUND",
            message=f"HRIS 集成 {hris_instance_id} 不存在或已禁用。",
            status_code=404,
        )
    config: dict = dict(row.config_public or {})
    if row.secrets_encrypted and row.secrets_nonce:
        config.update(decrypt_fields(
            row.secrets_encrypted,
            row.secrets_nonce,
            integration_id=str(row.id),
        ))
    return [SuccessFactorsLifecycleService.for_instance(
        db, integration_id=row.id, config=config,
    )]


def _serialize_batch(batch) -> SyncBatchResponse:
    """Build the full batch payload including staged events + sync counts.

    Centralised so the three endpoints that return a batch detail
    (preview window / preview full / incremental / get) stay consistent
    as we add fields.
    """
    # Direct attribute access on real ORM columns. ``getattr(row,
    # "col", default)`` masks rename-typo bugs (default silently
    # returned instead of AttributeError). Same bug class as PRs
    # 122/127/128/129. Every field below is a real column on
    # ``HrSyncBatch``.
    return SyncBatchResponse(
        id=batch.id,
        triggered_by=batch.triggered_by,
        trigger_source=batch.trigger_source,
        sync_mode=batch.sync_mode,
        is_seed=batch.is_seed,
        hris_instance_id=batch.hris_instance_id,
        start_date=batch.start_date,
        end_date=batch.end_date,
        since_datetime=batch.since_datetime,
        until_datetime=batch.until_datetime,
        status=batch.status,
        events_staged=batch.events_staged,
        new_count=batch.new_count,
        updated_count=batch.updated_count,
        reactivated_count=batch.reactivated_count,
        deactivated_count=batch.deactivated_count,
        skipped_count=batch.skipped_count,
        error_count=batch.error_count,
        error_message=batch.error_message,
        created_at=batch.created_at,
        reviewed_at=batch.reviewed_at,
        reviewed_by=batch.reviewed_by,
        staged_events=[
            StagedEventResponse(
                id=e.id,
                batch_id=e.batch_id,
                event_type=e.event_type,
                user_id=e.user_id,
                person_data=e.person_data,
                employment_data=e.employment_data,
                action=e.action,
                existing_person_id=e.existing_person_id,
                status=e.status,
                created_at=e.created_at,
            )
            for e in (batch.staged_events or [])
        ],
    )


@router.post(
    "/successfactors/test-connection",
    summary="Test SuccessFactors OData connectivity with the currently-saved platform config",
)
async def test_successfactors_connection(
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """One-shot connectivity probe.

    Reads the saved ``successfactors`` platform config, constructs the
    configured auth strategy, and performs a minimal
    ``FOCompany?$top=1`` request. Returns ``ok: True`` + a sample
    company name on success; surfaces a structured error (HTTP status,
    response snippet, or config gap) on failure so the admin can fix
    without digging into logs.
    """
    from app.identity.successfactors_client import SuccessFactorsClient

    audit = AuditService(db)
    trace_id = request.state.trace_id
    try:
        client = await SuccessFactorsClient.from_config(session=db)
        result = await client.ping()
    except AppException as e:
        await audit.log(
            trace_id=trace_id,
            actor_user_id=current_user.id,
            actor_type="user",
            action="hr.successfactors.test_connection",
            resource_type="platform_config",
            resource_id="successfactors",
            status="failure",
            detail={"error": e.code, "message": e.message},
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("user-agent"),
        )
        await db.commit()
        raise
    except SuccessFactorsHTTPError as e:
        # ``SuccessFactorsClient._get_json`` now raises this typed error
        # with the parsed OData ``error.code`` + ``error.message.value``
        # already extracted, so the admin sees the SF-side reason
        # ("Property 'positionNameEN' is invalid", etc.) rather than a
        # generic ``HTTP 400``.
        payload = {
            "ok": False,
            "http_status": e.status_code,
            "error": e.detail,
        }
        await audit.log(
            trace_id=trace_id,
            actor_user_id=current_user.id,
            actor_type="user",
            action="hr.successfactors.test_connection",
            resource_type="platform_config",
            resource_id="successfactors",
            status="failure",
            detail=payload,
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("user-agent"),
        )
        await db.commit()
        return success_response(data=payload, request=request)
    except Exception as e:  # noqa: BLE001
        public_payload = {
            "ok": False,
            "error": "CONNECTION_TEST_FAILED",
        }
        await audit.log(
            trace_id=trace_id,
            actor_user_id=current_user.id,
            actor_type="user",
            action="hr.successfactors.test_connection",
            resource_type="platform_config",
            resource_id="successfactors",
            status="failure",
            detail={
                **public_payload,
                "error_type": type(e).__name__,
            },
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("user-agent"),
        )
        await db.commit()
        logger.error(
            "successfactors connection test failed trace_id=%s error_type=%s",
            trace_id,
            type(e).__name__,
        )
        return success_response(data=public_payload, request=request)

    await audit.log(
        trace_id=trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.successfactors.test_connection",
        resource_type="platform_config",
        resource_id="successfactors",
        status="success",
        detail={"auth": result.get("auth", {}).get("type")},
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post(
    "/sync/full-employees/preview",
    summary="Preview a full-employee sync — fan out across every enabled HRIS instance",
)
async def preview_full_employee_sync(
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
    hris_instance_id: UUID | None = Query(
        default=None,
        description="仅对指定实例做全量同步(省略时对所有已启用的 HRIS 实例逐一执行)。",
    ),
    as_seed: bool | None = Query(
        default=None,
        description=(
            "是否按「初始播种」处理。播种批次会抑制全部生命周期协调器"
            "(不建号、不开离职流程)并豁免数量刹车 —— 只有一个实例第一次"
            "整册入库时才该这样。省略时由平台判断(该实例既无人员也无生命"
            "周期事件才算播种)。例行全量对账必须是 false,否则它捞回来的"
            "组织变动/新人/离职会被静默丢弃。"
        ),
    ),
):
    """Full-scan companion to :func:`preview_sync`.

    For each enabled HRIS integration (CN + EU + …), pulls every
    ``status eq 't'`` User and stages an ``onboard`` event so the
    batch-review flow can seed every active employee. Returns one
    batch per instance so the admin can confirm them individually —
    e.g. review CN first, then EU.

    Pass ``hris_instance_id`` to target a single region; omit to
    fan out across all enabled instances.

    ``as_seed`` (migration 565) overrides the seed predicate. It is a
    query parameter rather than an inference because a deliberate
    historical re-import is a decision a human makes; before it existed,
    the word "full" made that decision for them and every weekly
    reconciliation inherited seed semantics.
    """
    from sqlalchemy import select as _select
    from sqlalchemy.orm import selectinload

    from app.domains.hr.models import HrSyncBatch

    services = await _resolve_services(db, hris_instance_id=hris_instance_id)

    batches: list = []
    errors: list[dict] = []
    for svc in services:
        async with best_effort(
            db, label="hr: preview full employee sync per instance", log=logger,
        ) as full_preview_step:
            batch = await svc.preview_full_employee_sync(
                triggered_by=str(current_user.id),
                is_seed=as_seed,
            )
            batch = await db.scalar(
                _select(HrSyncBatch)
                .options(selectinload(HrSyncBatch.staged_events))
                .where(HrSyncBatch.id == batch.id)
            )
            batches.append(batch)
        if not full_preview_step.ok:
            errors.append({
                "hris_instance_id": str(svc.hris_instance_id) if svc.hris_instance_id else None,
                "error": "HRIS_PREVIEW_FAILED",
                "error_type": full_preview_step.error_type,
            })

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.sync.full_employees.preview",
        resource_type="hr_sync_batch",
        resource_id=",".join(str(b.id) for b in batches) or "fanout",
        status="success" if not errors else "partial",
        detail={
            "batch_ids": [str(b.id) for b in batches],
            "total_events_staged": sum(b.events_staged for b in batches),
            "instance_count": len(services),
            "errors": errors,
        },
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    payload = {
        "batches": [_serialize_batch(b).model_dump(mode="json") for b in batches],
        "errors": errors,
    }
    return success_response(data=payload, request=request)


@router.post("/sync/preview", summary="Preview HRIS window sync — fan out across every enabled HRIS instance")
async def preview_sync(
    body: SyncPreviewRequest,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
    hris_instance_id: UUID | None = Query(
        default=None,
        description="仅对指定实例同步窗口内的入离职事件(省略时对所有已启用的 HRIS 实例逐一执行)。",
    ),
):
    """Fetch hire/termination events in ``[start_date, end_date]`` from
    each HRIS instance, compute diff, stage batches for admin review.
    Returns one batch per instance; errors on one instance don't
    block the others.
    """
    from sqlalchemy import select as _select
    from sqlalchemy.orm import selectinload

    from app.domains.hr.models import HrSyncBatch

    start_date = body.start_date or date.today()
    end_date = body.end_date or start_date

    services = await _resolve_services(db, hris_instance_id=hris_instance_id)

    batches: list = []
    # Keep each successful batch paired with the service that produced
    # it — an errored instance appends to ``errors`` (not ``batches``),
    # so ``zip(services, batches)`` would misalign and auto-confirm one
    # instance's batch through another instance's (wrong-tenant-scoped)
    # service. Pair at success time instead.
    batch_svc_pairs: list[tuple] = []
    errors: list[dict] = []
    for svc in services:
        async with best_effort(
            db, label="hr: preview import per instance", log=logger,
        ) as preview_step:
            batch = await svc.preview_import(
                start_date=start_date,
                end_date=end_date,
                triggered_by=str(current_user.id),
            )
            batch = await db.scalar(
                _select(HrSyncBatch)
                .options(selectinload(HrSyncBatch.staged_events))
                .where(HrSyncBatch.id == batch.id)
            )
            batches.append(batch)
            batch_svc_pairs.append((svc, batch))
        if not preview_step.ok:
            errors.append({
                "hris_instance_id": str(svc.hris_instance_id) if svc.hris_instance_id else None,
                "error": "HRIS_PREVIEW_FAILED",
                "error_type": preview_step.error_type,
            })

    # HRIS small-delta auto-confirm (2026-06-01). After preview, any
    # batch whose delta is small + low-risk (new-hires ≤ threshold, no
    # terminations, no org moves — gated by platform_settings, OFF by
    # default) is confirmed immediately + a Feishu summary card is sent.
    # Each batch is evaluated independently; one ineligible batch never
    # blocks another. The service writes its own audit + commits.
    auto_confirm_verdicts: list[dict] = []
    for svc, batch in batch_svc_pairs:
        # A savepoint around a block that commits mid-flight is released
        # by that commit, so the block exit operates on a freed one and
        # every batch reports as failed — one of the three shapes
        # PLATFORM.md #25 names as must-not-wrap.
        # savepoint-exempt: auto_confirm_if_eligible commits per batch
        try:
            verdict = await svc.auto_confirm_if_eligible(
                batch,
                actor_id=str(current_user.id),
                trace_id=str(request.state.trace_id),
            )
            auto_confirm_verdicts.append({
                "batch_id": str(batch.id),
                "confirmed": verdict.get("confirmed", False),
                "reason": verdict.get("reason"),
            })
        except Exception as exc:  # noqa: BLE001 — never block preview
            auto_confirm_verdicts.append({
                "batch_id": str(batch.id),
                "confirmed": False,
                "reason": f"error:{type(exc).__name__}",
            })

    # Re-fetch so the response reflects any auto-confirmed status.
    if batches:
        batch_ids = [b.id for b in batches]
        refreshed = (await db.execute(
            _select(HrSyncBatch)
            .options(selectinload(HrSyncBatch.staged_events))
            .where(HrSyncBatch.id.in_(batch_ids))
        )).scalars().all()
        by_id = {b.id: b for b in refreshed}
        batches = [by_id.get(bid, b) for bid, b in zip(batch_ids, batches)]

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.sync.preview",
        resource_type="hr_sync_batch",
        resource_id=",".join(str(b.id) for b in batches) or "fanout",
        status="success" if not errors else "partial",
        detail={
            "start_date": start_date.isoformat(),
            "end_date": end_date.isoformat(),
            "batch_ids": [str(b.id) for b in batches],
            "total_events_staged": sum(b.events_staged for b in batches),
            "errors": errors,
            "auto_confirm": auto_confirm_verdicts,
        },
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    payload = {
        "batches": [_serialize_batch(b).model_dump(mode="json") for b in batches],
        "errors": errors,
        "auto_confirm": auto_confirm_verdicts,
    }
    return success_response(data=payload, request=request)


@router.get("/sync/batches", summary="List HRIS sync batches")
async def list_sync_batches(
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
    hris_instance_id: UUID | None = Query(
        default=None,
        description=(
            "Optional filter: scope the batch list to a single HRIS "
            "instance (multi-region setups). Omit to see all instances."
        ),
    ),
):
    service = SuccessFactorsLifecycleService(db)
    batches, total = await service.list_batches(
        page=page, page_size=page_size, hris_instance_id=hris_instance_id,
    )
    # Direct attribute access (see _build_batch_response above) — same
    # rationale: no getattr fallback on real ORM columns.
    payload = [
        SyncBatchSummaryResponse(
            id=b.id,
            triggered_by=b.triggered_by,
            trigger_source=b.trigger_source,
            sync_mode=b.sync_mode,
            is_seed=b.is_seed,
            hris_instance_id=b.hris_instance_id,
            start_date=b.start_date,
            end_date=b.end_date,
            since_datetime=b.since_datetime,
            until_datetime=b.until_datetime,
            status=b.status,
            events_staged=b.events_staged,
            new_count=b.new_count,
            updated_count=b.updated_count,
            reactivated_count=b.reactivated_count,
            deactivated_count=b.deactivated_count,
            skipped_count=b.skipped_count,
            error_count=b.error_count,
            created_at=b.created_at,
            reviewed_at=b.reviewed_at,
            reviewed_by=b.reviewed_by,
        ).model_dump(mode="json")
        for b in batches
    ]
    return success_response(
        data=payload,
        meta={"page": page, "page_size": page_size, "total": total},
        request=request,
    )


@router.post(
    "/sync/incremental/preview",
    summary="Preview an incremental HRIS sync (lastModifiedDateTime-based)",
)
async def preview_incremental_sync(
    body: IncrementalSyncRequest,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Run an on-demand incremental sync against one HRIS instance.

    Pulls users with ``lastModifiedDateTime >= since`` and stages the
    diff. ``since`` defaults to the end of the last successful batch
    (minus a safety margin) when the caller omits it, so a plain
    ``POST`` with an empty body does "pull what's new since last time"
    — the same path the scheduler uses.

    When ``auto_confirm=true`` the staged events are applied in the
    same request and the response's ``status`` will be ``confirmed``;
    otherwise the batch stays ``pending_review`` for manual review
    from /admin/user-lifecycle.
    """
    from datetime import UTC as _UTC
    from datetime import datetime as _datetime
    from datetime import timedelta as _timedelta

    from sqlalchemy import select as _select
    from sqlalchemy.orm import selectinload

    from app.domains.hr.models import HrSyncBatch
    from app.integrations.compat import list_hris_instances

    if body.hris_instance_id is None:
        instances = await list_hris_instances(session=db)
        if len(instances) == 0:
            raise AppException(
                code="INTEGRATION_NOT_CONFIGURED",
                message="没有已启用的 HRIS 集成。请先在 /admin/integrations 配置至少一个 SuccessFactors 实例。",
                status_code=503,
            )
        if len(instances) > 1:
            raise AppException(
                code="AMBIGUOUS_HRIS_INSTANCE",
                message="检测到多个 HRIS 实例,请在请求体中指定 hris_instance_id。",
                status_code=400,
            )
        instance_row, instance_config = instances[0]
    else:
        from sqlalchemy import select as _sel

        from app.domains.cmdb.credential_crypto import decrypt_fields
        from app.integrations.models import Integration
        instance_row = await db.scalar(
            _sel(Integration).where(
                Integration.id == body.hris_instance_id,
                Integration.kind == "hris",
            )
        )
        if instance_row is None or not instance_row.enabled:
            raise AppException(
                code="INTEGRATION_NOT_FOUND",
                message=f"HRIS 集成 {body.hris_instance_id} 不存在或已禁用。",
                status_code=404,
            )
        instance_config = dict(instance_row.config_public or {})
        if instance_row.secrets_encrypted and instance_row.secrets_nonce:
            instance_config.update(
                decrypt_fields(
                    instance_row.secrets_encrypted,
                    instance_row.secrets_nonce,
                    integration_id=str(instance_row.id),
                )
            )

    service = SuccessFactorsLifecycleService.for_instance(
        db,
        integration_id=instance_row.id,
        config=instance_config,
    )

    if body.since is None:
        last_until = await service._last_successful_until()
        since = (
            last_until - _timedelta(hours=1)
            if last_until
            else _datetime.now(_UTC) - _timedelta(hours=24)
        )
    else:
        since = body.since

    until = _datetime.now(_UTC)

    batch = await service.preview_incremental_sync(
        triggered_by=str(current_user.id),
        since=since,
        until=until,
        trigger_source="manual",
    )

    confirm_result: dict | None = None
    if body.auto_confirm:
        confirm_result = await service.confirm_batch(
            batch.id, confirmed_by=str(current_user.id),
        )

    batch = await db.scalar(
        _select(HrSyncBatch)
        .options(selectinload(HrSyncBatch.staged_events))
        .where(HrSyncBatch.id == batch.id)
    )

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.sync.incremental.preview",
        resource_type="hr_sync_batch",
        resource_id=str(batch.id),
        status="success",
        detail={
            "hris_instance_id": str(instance_row.id),
            "hris_instance_slug": instance_row.slug,
            "since": since.isoformat(),
            "until": until.isoformat(),
            "events_staged": batch.events_staged,
            "auto_confirmed": bool(confirm_result),
        },
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    payload = _serialize_batch(batch).model_dump(mode="json")
    if confirm_result is not None:
        payload["confirm_result"] = SyncConfirmResponse(**confirm_result).model_dump(mode="json")
    return success_response(data=payload, request=request)


@router.get(
    "/sync/instances",
    summary="List configured HRIS instances (for multi-region routing)",
)
async def list_hris_instances_endpoint(
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    """Return a compact view of every enabled HRIS integration.

    The admin UI uses this to offer a region selector when triggering
    sync actions; the scheduler uses the same helper internally. Only
    non-secret fields are returned.
    """
    from app.integrations.compat import list_hris_instances
    instances = await list_hris_instances(session=db)
    payload = [
        {
            "id": str(row.id),
            "slug": row.slug,
            "name": row.name,
            "enabled": row.enabled,
            "is_primary": row.is_primary,
            "region": (config.get("region") or None),
            "base_url": (config.get("base_url") or None),
            "auth_method": (config.get("auth_method") or None),
            "last_tested_at": row.last_tested_at,
            "last_test_status": row.last_test_status,
        }
        for row, config in instances
    ]
    return success_response(data=payload, request=request)


@router.get("/sync/batches/{batch_id}", summary="Get sync batch detail with staged events")
async def get_sync_batch(
    batch_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    service = SuccessFactorsLifecycleService(db)
    batch = await service.get_batch(batch_id)
    payload = _serialize_batch(batch)
    return success_response(data=payload.model_dump(mode="json"), request=request)


@router.post(
    "/sync/batches/{batch_id}/confirm",
    status_code=202,
    summary="Confirm a staged sync batch — applies events asynchronously",
)
async def confirm_sync_batch(
    batch_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Flip the batch to ``applying`` and dispatch the heavy per-event
    apply to a background task, returning ``202`` immediately.

    Why async (2026-06-16 prod incident): a full-bootstrap batch applies
    ~500 events plus org-wide HRBP/Budget-Owner refresh, manager-backfill
    and annual-leave tail passes. Run synchronously inside this request it
    took minutes — past the frontend's axios timeout, which aborted the
    connection (nginx logged ``499``) and showed the admin an error toast
    even though the apply actually completed. The work now runs off the
    request thread and the UI polls ``GET /sync/batches/{id}`` to terminal.

    The mass-leaver brake still runs synchronously inside
    ``start_async_confirm`` so its refusal surfaces as an immediate
    ``409`` rather than vanishing into the background task. A backend
    restart mid-apply is recovered by the ``hris.resume_stuck_apply`` cron.
    """
    service = SuccessFactorsLifecycleService(db)
    batch, did_flip = await service.start_async_confirm(
        batch_id, confirmed_by=str(current_user.id)
    )
    # Capture the FK now (the ORM object's attributes are read before we
    # hand control to the detached background session).
    hris_instance_id = batch.hris_instance_id

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.sync.confirm_started",
        resource_type="hr_sync_batch",
        resource_id=str(batch_id),
        status="success",
        detail={"status": batch.status, "events_staged": batch.events_staged},
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    # Optimistic immediate start (best UX); the resume cron is the durable
    # backstop if this process dies before the apply finishes. ``spawn``
    # retains a strong reference so the task isn't GC-dropped mid-flight —
    # the failure mode that silently lost every FastAPI BackgroundTask in
    # PR #279 (migration 239).
    #
    # #1743: spawn ONLY when THIS request performed the pending_review →
    # applying flip. A double-click / axios-timeout resend hits an
    # already-``applying`` batch (``did_flip=False``); spawning a second
    # runner there races the first on the same staged rows — the resulting
    # lock contention mislabels events 'failed' and the slower runner
    # overwrites Phase C's new/updated counts to ~0. The still-running
    # first runner (or the resume cron) already owns the apply, so this
    # request just returns the current batch state for the UI to poll.
    if did_flip:
        _spawn_background(
            run_batch_apply_in_background(
                batch_id,
                str(current_user.id),
                hris_instance_id,
            ),
            name=f"hris-apply-{batch_id}",
        )

    payload = _serialize_batch(await service.get_batch(batch_id)).model_dump(mode="json")
    return success_response(data=payload, request=request)


@router.post("/sync/batches/{batch_id}/cancel", summary="Cancel a pending sync batch")
async def cancel_sync_batch(
    batch_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = SuccessFactorsLifecycleService(db)
    await service.cancel_batch(batch_id, cancelled_by=str(current_user.id))

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.sync.cancel",
        resource_type="hr_sync_batch",
        resource_id=str(batch_id),
        status="success",
        detail={},
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    return success_response(data={"batch_id": str(batch_id), "status": "cancelled"}, request=request)


# ── HRIS small-delta auto-confirm: settings + history ──


class AutoConfirmSettingsBody(BaseModel):
    auto_confirm_small_delta: bool
    auto_confirm_threshold_count: int = Field(ge=0, le=1000)
    auto_confirm_block_on_termination: bool
    auto_confirm_block_on_manager_change: bool


@router.get(
    "/sync/auto-confirm/settings",
    summary="Read HRIS small-delta auto-confirm settings",
)
async def get_auto_confirm_settings(
    request: Request,
    _admin: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.hr.service import (
        AUTO_CONFIRM_DEFAULT_THRESHOLD,
        SETTING_AUTO_CONFIRM_BLOCK_MANAGER,
        SETTING_AUTO_CONFIRM_BLOCK_TERMINATION,
        SETTING_AUTO_CONFIRM_ENABLED,
        SETTING_AUTO_CONFIRM_THRESHOLD,
    )
    from app.integrations.service import get_platform_setting

    data = {
        "auto_confirm_small_delta": bool(await get_platform_setting(
            SETTING_AUTO_CONFIRM_ENABLED, default=False, session=db)),
        "auto_confirm_threshold_count": int(await get_platform_setting(
            SETTING_AUTO_CONFIRM_THRESHOLD,
            default=AUTO_CONFIRM_DEFAULT_THRESHOLD, session=db) or 0),
        "auto_confirm_block_on_termination": bool(await get_platform_setting(
            SETTING_AUTO_CONFIRM_BLOCK_TERMINATION, default=True, session=db)),
        "auto_confirm_block_on_manager_change": bool(await get_platform_setting(
            SETTING_AUTO_CONFIRM_BLOCK_MANAGER, default=True, session=db)),
    }
    return success_response(data=data, request=request)


@router.put(
    "/sync/auto-confirm/settings",
    summary="Update HRIS small-delta auto-confirm settings",
)
async def update_auto_confirm_settings(
    body: AutoConfirmSettingsBody,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.hr.service import (
        SETTING_AUTO_CONFIRM_BLOCK_MANAGER,
        SETTING_AUTO_CONFIRM_BLOCK_TERMINATION,
        SETTING_AUTO_CONFIRM_ENABLED,
        SETTING_AUTO_CONFIRM_THRESHOLD,
    )
    from app.integrations.service import PlatformSettingService

    svc = PlatformSettingService(db)
    trace_id = str(request.state.trace_id)
    pairs = [
        (SETTING_AUTO_CONFIRM_ENABLED, body.auto_confirm_small_delta,
         "HRIS small-delta auto-confirm master switch"),
        (SETTING_AUTO_CONFIRM_THRESHOLD, body.auto_confirm_threshold_count,
         "Max new-hire count eligible for auto-confirm"),
        (SETTING_AUTO_CONFIRM_BLOCK_TERMINATION,
         body.auto_confirm_block_on_termination,
         "Block auto-confirm when the batch contains terminations"),
        (SETTING_AUTO_CONFIRM_BLOCK_MANAGER,
         body.auto_confirm_block_on_manager_change,
         "Block auto-confirm on manager/department (org-line) changes"),
    ]
    for key, value, desc in pairs:
        await svc.upsert(
            key, value, desc, trace_id=trace_id,
            actor_user_id=current_user.id,
        )
    await db.commit()
    return success_response(data=body.model_dump(), request=request)


@router.get(
    "/sync/auto-confirm/history",
    summary="Recent HRIS small-delta auto-confirm runs (last 30 days)",
)
async def list_auto_confirm_history(
    request: Request,
    _admin: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.audit.models import AuditLog
    from app.domains.hr.service import AUDIT_ACTION_AUTO_CONFIRMED

    since = datetime.now(UTC) - timedelta(days=30)
    rows = (await db.execute(
        select(AuditLog)
        .where(
            AuditLog.action == AUDIT_ACTION_AUTO_CONFIRMED,
            AuditLog.created_at >= since,
        )
        .order_by(AuditLog.created_at.desc())
        .limit(200)
    )).scalars().all()
    items = [
        {
            "batch_id": r.resource_id,
            "created_at": r.created_at.isoformat() if r.created_at else None,
            "detail": r.detail,
        }
        for r in rows
    ]
    return success_response(data={"items": items}, request=request)


@router.post("/successfactors/webhook")
async def receive_successfactors_webhook_primary(
    request: Request,
    db: AsyncSession = Depends(get_db),
    webhook_token: str | None = Header(default=None, alias="X-Helix-Webhook-Token"),
    _rate: None = Depends(check_login_rate_limit),
):
    """Primary-instance webhook (back-compat URL).

    Kept for single-instance installs and older SF configurations.
    Multi-region setups should point SF at the slug-scoped URL
    ``/successfactors/webhook/{slug}`` instead so each instance's
    events are attributed to the right region.
    """
    return await _handle_sf_webhook(
        db=db, request=request,
        webhook_token=webhook_token, integration_slug=None,
    )


# ── Diagnose routes declared BEFORE the {integration_slug} catch-all ──
# FastAPI matches routes in declaration order; without this ordering
# the slug route swallows ``/successfactors/webhook/_diagnose`` (treats
# ``_diagnose`` as a literal slug) and the diagnose endpoint silently
# 404s.
@router.post("/successfactors/webhook/_diagnose")
async def diagnose_successfactors_webhook_primary(
    request: Request,
    db: AsyncSession = Depends(get_db),
    webhook_token: str | None = Header(default=None, alias="X-Helix-Webhook-Token"),
    _rate: None = Depends(check_login_rate_limit),
):
    """Dry-run validate an SF webhook request: same auth envelope as
    the real endpoint, but DO NOT apply any events. HRIS integrators
    configuring the SF Integration Center / CPI iflow point at this
    endpoint first to verify token + HMAC signature + payload schema
    before flipping their connector to the apply endpoint.

    Returns a per-event diagnostic:

    * ``user_id`` — copied from the request
    * ``event_type`` — copied from the request
    * ``person_match`` — ``{person_id, display_name, employment_status,
      hire_date, termination_date}`` if Helix already has a person row
      for this ``hris_person_id`` under the SF source-system; ``null``
      otherwise (likely OK for an onboard, suspicious for an offboard).

    No DB writes; the response is purely informational so configuration
    issues surface immediately without polluting production data.
    """
    return await _handle_sf_webhook_diagnose(
        db=db, request=request,
        webhook_token=webhook_token, integration_slug=None,
    )


@router.post("/successfactors/webhook/{integration_slug}/_diagnose")
async def diagnose_successfactors_webhook_scoped(
    integration_slug: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
    webhook_token: str | None = Header(default=None, alias="X-Helix-Webhook-Token"),
    _rate: None = Depends(check_login_rate_limit),
):
    """Instance-scoped diagnose URL. Mirrors
    :func:`receive_successfactors_webhook_scoped` so multi-region
    integrators can verify each iflow against the slug it will
    eventually post to. No DB writes.
    """
    return await _handle_sf_webhook_diagnose(
        db=db, request=request,
        webhook_token=webhook_token, integration_slug=integration_slug,
    )


@router.post("/successfactors/webhook/{integration_slug}")
async def receive_successfactors_webhook_scoped(
    integration_slug: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
    webhook_token: str | None = Header(default=None, alias="X-Helix-Webhook-Token"),
    _rate: None = Depends(check_login_rate_limit),
):
    """Instance-scoped webhook — SF configures this URL per region so
    incoming events get tagged with the correct ``hris_instance_id``.

    Example URLs:

        POST /api/v1/hr/successfactors/webhook/sf-cn-dev
        POST /api/v1/hr/successfactors/webhook/sf-eu-dev
    """
    return await _handle_sf_webhook(
        db=db, request=request,
        webhook_token=webhook_token, integration_slug=integration_slug,
    )


async def _handle_sf_webhook_diagnose(
    *,
    db: AsyncSession,
    request: Request,
    webhook_token: str | None,
    integration_slug: str | None,
) -> object:
    from app.identity.person_models import IdentityPerson

    integration_row, _sf_config, body = await _authenticate_sf_webhook_request(
        db=db, request=request,
        webhook_token=webhook_token, integration_slug=integration_slug,
    )

    user_ids = sorted({event.user_id for event in body.events if event.user_id})
    persons_by_hris_id: dict[str, IdentityPerson] = {}
    cross_instance_ids: set[str] = set()
    if user_ids:
        stmt = select(IdentityPerson).where(
            IdentityPerson.source_system == "successfactors",
            IdentityPerson.hris_person_id.in_(user_ids),
        )
        if integration_row is not None:
            # #2235: legacy rows predate ``hris_instance_id`` and carry
            # NULL. Every other instance-scoped lookup (service.py's
            # ``_scoped_person_query`` family) admits them with the same
            # ``or_(... .is_(None))`` clause, and the apply path treats
            # them as belonging to this tenant — without the clause here
            # diagnose pushed them into the cross-tenant fallback and
            # mislabelled them ``cross_instance: true``, i.e. the
            # diagnostic disagrees with what apply actually does.
            stmt = stmt.where(
                or_(
                    IdentityPerson.hris_instance_id == integration_row.id,
                    IdentityPerson.hris_instance_id.is_(None),
                )
            )
        result = await db.execute(stmt)
        for person in result.scalars():
            persons_by_hris_id[person.hris_person_id] = person
        # Cross-tenant fallback (2026-07-31, P01550 cohort): terminations
        # are frequently entered in the tenant that does NOT own the
        # Helix row (EU approval for a CN-pinned person), and the apply
        # path adopts those on offboard (see ``_upsert_person``'s
        # offboard adoption branch). Diagnose must mirror that
        # resolution or the integrator's smoke test reports
        # ``resolved_persons: 0`` for pushes that WOULD apply — exactly
        # what sent HRIS's first real test into triage. Flagged
        # ``cross_instance`` so a genuinely mis-routed push is still
        # visible.
        missing = [u for u in user_ids if u not in persons_by_hris_id]
        if missing and integration_row is not None:
            fallback = await db.execute(
                # ``deleted_at IS NULL`` (#2235): the apply path never
                # resolves tombstones, so neither may the diagnostic —
                # otherwise a purged person (rogerzhou02 shape) is
                # reported ``resolved`` and the smoke test blesses a push
                # that will in fact create a fresh row.
                select(IdentityPerson).where(
                    IdentityPerson.source_system == "successfactors",
                    IdentityPerson.hris_person_id.in_(missing),
                    IdentityPerson.deleted_at.is_(None),
                )
            )
            for person in fallback.scalars():
                persons_by_hris_id[person.hris_person_id] = person
                cross_instance_ids.add(person.hris_person_id)

    # P1-C: diagnose returns only the resolution boolean + employment_status —
    # NOT display_name / work_email / hire_date / termination_date. Anyone
    # who holds the webhook secret + a captured timestamp can otherwise
    # enumerate `hris_person_id` ranges via diagnose and dump the full
    # directory (name + work email + hire/termination dates). The apply
    # endpoint doesn't echo person attributes at all; diagnose now matches
    # that minimum-disclosure envelope. ``person_id`` (a non-enumerable
    # UUID) is kept so the integrator can cross-reference back to a
    # specific Helix row when triaging unresolved cases.
    events_diag: list[dict] = []
    for event in body.events:
        person = persons_by_hris_id.get(event.user_id)
        match: dict | None = None
        if person is not None:
            match = {
                "person_id": str(person.id),
                "employment_status": person.employment_status,
            }
            if event.user_id in cross_instance_ids:
                # Resolved via the cross-tenant fallback — the person is
                # currently pinned to another HRIS instance. Offboards
                # still apply (adopted); see _upsert_person.
                match["cross_instance"] = True
        events_diag.append({
            "user_id": event.user_id,
            "event_type": event.event_type,
            "effective_date": event.effective_date.isoformat(),
            "person_id_external": event.person_id_external,
            "person_match": match,
            "would_apply": event.event_type in {"onboard", "offboard"},
        })

    payload = {
        "signature_ok": True,
        "schema_ok": True,
        "events_received": len(body.events),
        "resolved_persons": len(persons_by_hris_id),
        "unresolved_user_ids": [
            uid for uid in user_ids if uid not in persons_by_hris_id
        ],
        "events": events_diag,
        "hris_instance_slug": (
            integration_row.slug if integration_row else None
        ),
    }

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_type="system",
        action="hr.successfactors_webhook.diagnose",
        resource_type="hr_lifecycle_event",
        resource_id=str(len(body.events)),
        status="success",
        detail={
            "events_received": payload["events_received"],
            "resolved_persons": payload["resolved_persons"],
            "unresolved_user_ids_count": len(payload["unresolved_user_ids"]),
            "hris_instance_slug": payload["hris_instance_slug"],
        },
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    return success_response(data=payload, request=request)


def _ip_in_allowlist(client_ip: str, allowlist: list) -> bool:
    """True if ``client_ip`` falls inside any IP/CIDR in ``allowlist``.

    A non-IP client value (e.g. ``"unknown"`` when the peer can't be
    resolved) fails closed — when an allowlist is set, an unresolvable
    caller is rejected. Malformed allowlist entries are skipped.
    """
    import ipaddress

    try:
        ip = ipaddress.ip_address(client_ip)
    except ValueError:
        return False
    for entry in allowlist:
        try:
            if ip in ipaddress.ip_network(str(entry).strip(), strict=False):
                return True
        except ValueError:
            continue
    return False


def _parse_sf_webhook_body(raw_body: bytes) -> SuccessFactorsWebhookRequest:
    """Parse the webhook JSON, accepting the shapes upstream tools actually
    emit (be liberal in what we accept):

    * ``{"events": [ {...} ]}``  — canonical wrapper
    * ``[ {...}, {...} ]``       — a bare array of event objects. SAP
      SuccessFactors Integration Center, with a record-level field mapping
      (root fields ``event_type`` / ``user_id`` / …), POSTs exactly this —
      a top-level JSON array, not the wrapper. Forcing the integrator to
      build a nested ``events`` node is the whole reason this normalisation
      exists.
    * ``{...}``                  — a single bare event object
    * NDJSON: one ``{...}`` per line — some connector configs

    Raises ``AppException(WEBHOOK_BODY_INVALID, 400)`` on anything else.
    """
    text = (raw_body or b"").decode("utf-8", errors="replace").strip()
    if not text:
        data: object = {"events": []}
    else:
        try:
            data = json.loads(text)
        except json.JSONDecodeError:
            # Fallback: newline-delimited JSON objects. Cap the line
            # count so a (necessarily authenticated) caller can't drive
            # tens of millions of json.loads with a max-size body of
            # single-char lines — a real SF sync batch is well under 10k
            # events. Defence-in-depth on top of nginx's body-size limit.
            rows: list = []
            try:
                for line in text.splitlines():
                    line = line.strip()
                    if line:
                        if len(rows) >= 10000:
                            raise AppException(
                                code="WEBHOOK_BODY_INVALID",
                                message=(
                                    "Webhook NDJSON body exceeds the 10000-event "
                                    "limit; split into smaller batches."
                                ),
                                status_code=400,
                            )
                        rows.append(json.loads(line))
            except json.JSONDecodeError as exc:
                raise AppException(
                    code="WEBHOOK_BODY_INVALID",
                    message=f"Webhook body is not valid JSON: {exc}",
                    status_code=400,
                ) from exc
            data = rows

    if isinstance(data, list):
        payload: dict = {"events": data}
    elif isinstance(data, dict):
        # Has the wrapper → use as-is; otherwise treat the object itself as
        # a single event (empty object ⇒ no events, not a one-bad-event).
        payload = data if "events" in data else (
            {"events": [data]} if data else {"events": []}
        )
    else:
        raise AppException(
            code="WEBHOOK_BODY_INVALID",
            message="Webhook body must be a JSON object or array of events.",
            status_code=400,
        )

    try:
        return SuccessFactorsWebhookRequest.model_validate(payload)
    except ValidationError as exc:
        raise AppException(
            code="WEBHOOK_BODY_INVALID",
            message=f"Webhook body parse error: {exc}",
            status_code=400,
        ) from exc


async def _authenticate_sf_webhook_request(
    *,
    db: AsyncSession,
    request: Request,
    webhook_token: str | None,
    integration_slug: str | None,
):
    """Validate token + replay window + body HMAC signature for an SF
    webhook request, parse the body, and resolve the owning integration
    row and SF config.

    Returns ``(integration_row, sf_config, body)``. Raises the same
    auth / validation errors callers expect — never returns a partial
    state. Shared between the real apply endpoint and the diagnose
    endpoint so the security envelope is identical (a passing
    diagnostic IS a passing real call, modulo side-effects).
    """
    from app.domains.cmdb.credential_crypto import decrypt_fields
    from app.integrations.compat import get_successfactors_config
    from app.integrations.models import Integration

    integration_row: Integration | None = None
    if integration_slug is not None:
        integration_row = await db.scalar(
            select(Integration).where(
                Integration.slug == integration_slug,
                Integration.kind == "hris",
            )
        )
        if integration_row is None or not integration_row.enabled:
            raise AppException(
                code="INTEGRATION_NOT_FOUND",
                message=f"HRIS 集成 slug={integration_slug!r} 不存在或已禁用。",
                status_code=404,
            )
        sf_config: dict = dict(integration_row.config_public or {})
        if integration_row.secrets_encrypted and integration_row.secrets_nonce:
            sf_config.update(decrypt_fields(
                integration_row.secrets_encrypted,
                integration_row.secrets_nonce,
                integration_id=str(integration_row.id),
            ))
    else:
        sf_config = await get_successfactors_config(session=db)

    # ── Source-IP allowlist (applies to EVERY auth scheme) ──
    # When configured, only callers whose resolved client IP falls inside
    # one of the ranges may proceed. This is what recovers most of the
    # security lost by allowing the simpler Basic scheme below: a leaked
    # Basic credential is useless from an IP outside the SF/CPI egress
    # ranges. An empty/absent allowlist means "allow all" (the admin's
    # explicit default — auth then rests on Basic/HMAC alone).
    ip_allowlist = sf_config.get("webhook_ip_allowlist") or []
    if ip_allowlist:
        client_ip = get_real_ip(request)
        if not _ip_in_allowlist(client_ip, ip_allowlist):
            raise AppException(
                code="WEBHOOK_IP_NOT_ALLOWED",
                message="调用方 IP 不在该集成的 webhook 白名单内。",
                status_code=403,
            )

    raw_body = await request.body()

    # ── Auth scheme A: HTTP Basic (simple path for zero-code upstream
    # tools) ──. SF Integration Center / similar connectors cannot compute
    # a per-request HMAC over the body; they CAN do Basic. If the caller
    # sends an ``Authorization: Basic`` header we validate it here and
    # skip the timestamp + signature requirement. HMAC stays available as
    # an alternative for callers that can sign (see scheme B). No replay
    # window on this path — the trade-off the IP allowlist + mass-leaver
    # brake + TLS are there to cover.
    auth_header = request.headers.get("Authorization") or ""
    if auth_header.startswith("Basic "):
        import base64 as _b64

        basic_user = (sf_config.get("webhook_basic_username") or "").strip()
        basic_pass = (sf_config.get("webhook_basic_password") or "").strip()
        if not basic_pass:
            raise UnauthorizedError(
                "Basic auth is not configured for this integration"
            )
        try:
            decoded = _b64.b64decode(auth_header[6:].strip()).decode("utf-8")
        except (ValueError, UnicodeDecodeError) as exc:
            raise UnauthorizedError("Malformed Basic auth header") from exc
        sent_user, _sep, sent_pass = decoded.partition(":")
        user_ok = (
            hmac.compare_digest(basic_user.encode(), sent_user.encode())
            if basic_user
            else True
        )
        pass_ok = hmac.compare_digest(basic_pass.encode(), sent_pass.encode())
        if not (user_ok and pass_ok):
            raise UnauthorizedError("Invalid Basic auth credentials")
        body = _parse_sf_webhook_body(raw_body)
        return integration_row, sf_config, body

    # ── Auth scheme B: token + replay window + body HMAC signature ──
    primary_secret = (sf_config.get("webhook_hmac_secret") or "").strip()
    if not primary_secret:
        # No HMAC secret configured. If Basic IS configured the caller
        # simply didn't send it — that's a 401, not a server misconfig.
        if (sf_config.get("webhook_basic_password") or "").strip():
            raise UnauthorizedError(
                "Missing credentials: send HTTP Basic auth or an "
                "HMAC-signed request"
            )
        raise AppException(
            code="WEBHOOK_TOKEN_NOT_CONFIGURED",
            message=(
                "SuccessFactors webhook secret 未配置。请在对应 HRIS 集成设置中填写 "
                "Webhook HMAC Secret(或点击「轮换 webhook 密钥」生成)。"
            ),
            status_code=500,
        )
    # Rolling-overlap dual-secret: during a rotation the previous
    # secret stays valid until ``webhook_hmac_secret_previous_expires_at``
    # so SF / CPI can roll into the new secret without dropping events.
    # Both secrets are HMAC-verified against the same algorithm; only
    # the candidate key set differs.
    candidate_secrets: list[tuple[str, str]] = [("primary", primary_secret)]
    previous_secret = (
        sf_config.get("webhook_hmac_secret_previous") or ""
    ).strip()
    previous_expires_at = sf_config.get("webhook_hmac_secret_previous_expires_at")
    if previous_secret and previous_expires_at:
        try:
            exp_dt = datetime.fromisoformat(str(previous_expires_at))
            if exp_dt.tzinfo is None:
                exp_dt = exp_dt.replace(tzinfo=UTC)
            if exp_dt > datetime.now(UTC):
                candidate_secrets.append(("previous", previous_secret))
        except ValueError:
            # Malformed expiry — treat as expired, never accept.
            pass

    if webhook_token is None or not any(
        hmac.compare_digest(s.encode(), webhook_token.encode())
        for _, s in candidate_secrets
    ):
        raise UnauthorizedError("Invalid webhook token")

    # Replay window: timestamp header MANDATORY. Pre-launch hardening
    # 2026-05-06 — the prior implementation only checked the window
    # when the header was present, so an attacker who once captured a
    # real SF webhook body+token could replay it indefinitely simply
    # by stripping the timestamp header.
    sf_timestamp_header = (
        request.headers.get("X-SF-Timestamp")
        or request.headers.get("X-Timestamp")
    )
    if not sf_timestamp_header:
        raise UnauthorizedError(
            "Webhook request missing replay-protection timestamp header"
        )
    try:
        import time as _time
        # Auto-detect ms vs s — SF deployments differ. Anything past
        # the year-5000 mark in seconds is interpreted as ms and
        # divided. Pre-launch P0 fix; previously a ms timestamp would
        # always be judged "too old" and the webhook silently 401'd.
        raw_ts = int(sf_timestamp_header)
        ts_seconds = raw_ts / 1000.0 if raw_ts > 1e11 else float(raw_ts)
        ts_age = abs(_time.time() - ts_seconds)
    except (ValueError, TypeError) as exc:
        raise UnauthorizedError(
            "Webhook request has malformed timestamp header"
        ) from exc
    if ts_age > 300:
        raise UnauthorizedError(
            "Webhook request timestamp is too old (replay protection)"
        )

    # ── Body HMAC verification (pre-launch P0 fix) ───────────────
    # The legacy auth was token-only — anyone with the
    # ``webhook_hmac_secret`` could craft arbitrary payloads and have
    # them mass-flip employees to ``leaver_pending``. Now require a
    # signature over ``timestamp + "." + raw_body`` so a leaked secret
    # alone is not enough; the attacker also needs the raw body bytes
    # to forge a valid signature, AND the signed payload remains
    # tied to a specific 5-minute window.
    raw_body = await request.body()
    sig_header = (
        request.headers.get("X-SF-Signature")
        or request.headers.get("X-Helix-Webhook-Signature")
    )
    if not sig_header:
        raise UnauthorizedError(
            "Webhook request missing X-SF-Signature header"
        )
    # Strip optional algorithm prefix ("sha256=...").
    sig_value = (
        sig_header.split("=", 1)[1] if "=" in sig_header else sig_header
    ).strip()

    # Verify against EVERY candidate secret (primary + still-valid
    # previous). Use a hmac.compare_digest pass per candidate; bail
    # the moment one matches so the timing-attack envelope is bounded
    # by the candidate count (2 max).
    matched_secret_kind: str | None = None
    for kind, secret in candidate_secrets:
        expected_signature = hmac.new(
            secret.encode(),
            f"{sf_timestamp_header}.".encode() + raw_body,
            hashlib.sha256,
        ).hexdigest()
        if hmac.compare_digest(expected_signature, sig_value):
            matched_secret_kind = kind
            break
    if matched_secret_kind is None:
        raise UnauthorizedError("Invalid webhook signature")
    if matched_secret_kind == "previous":
        # Surface in the worker log so an operator monitoring a
        # rotation can confirm SF/CPI has caught up. Not audited
        # per-call (audit row already records every accepted webhook).
        logger.info(
            "SF webhook accepted via previous (rotating) secret — "
            "integration_slug=%s expires_at=%s. SF/CPI may not yet "
            "have picked up the new secret.",
            integration_slug, previous_expires_at,
        )

    # Now safe to parse the body — signature verified.
    body = _parse_sf_webhook_body(raw_body)

    # P1-B: when accepted via the previous (rotating) secret, refuse
    # offboard events — narrows the attacker exfil window. An attacker
    # who captured the OLD secret + a recent real body has the 24h
    # overlap to forge ``leaver_pending`` floods; restricting the
    # previous-secret path to non-destructive events caps the blast
    # radius. SF / CPI legitimately catching up sees no impact for
    # onboard events (they keep flowing); for the rare case where a
    # real offboard arrives during the overlap window the iflow's
    # at-least-once retry will deliver it once SF/CPI completes the
    # rotate.
    if matched_secret_kind == "previous":
        offboard_events = [
            e for e in body.events if e.event_type == "offboard"
        ]
        if offboard_events:
            logger.warning(
                "SF webhook accepted via previous secret refused %d "
                "offboard event(s) — rotation-overlap policy. "
                "integration_slug=%s. iflow at-least-once retry will "
                "deliver these once SF/CPI picks up the new secret.",
                len(offboard_events), integration_slug,
            )
            raise UnauthorizedError(
                "Offboard events rejected during webhook-secret "
                "rotation overlap. Complete the rotation on the SF/CPI "
                "side and re-send."
            )

    return integration_row, sf_config, body


async def _log_sf_webhook_receipt(
    request: Request,
    *,
    slug: str,
    body: SuccessFactorsWebhookRequest,
) -> None:
    """Log only non-PII metadata for an authenticated SF webhook receipt."""
    try:
        raw_body = await request.body()
        body_bytes = len(raw_body)
        body_sha256 = hashlib.sha256(raw_body).hexdigest()
    except Exception:  # noqa: BLE001 — logging must never break the webhook
        # Do not log the exception: a request-body error may echo payload data.
        body_bytes = -1
        body_sha256 = "unavailable"

    # JSON string escaping prevents a malformed-but-valid event type from
    # injecting line breaks or forged fields into the receipt log.
    event_types = ",".join(
        json.dumps(event_type, ensure_ascii=True)[1:-1]
        for event_type in sorted({event.event_type for event in body.events})
    )
    logger.event(
        "HRIS_WEBHOOK_RECEIVED",
        {
            "trace_id": request.state.trace_id,
            "slug": slug,
            "event_types": event_types,
            "event_count": len(body.events),
            "body_bytes": body_bytes,
            "body_sha256": body_sha256,
        },
    )


async def _handle_sf_webhook(
    *,
    db: AsyncSession,
    request: Request,
    webhook_token: str | None,
    integration_slug: str | None,
) -> object:
    integration_row, sf_config, body = await _authenticate_sf_webhook_request(
        db=db, request=request,
        webhook_token=webhook_token, integration_slug=integration_slug,
    )

    # Fast-ACK design (504 fix, 2026-07-01): an SF onboard/offboard event
    # is just a dated signal, but ``process_webhook_events`` turns each
    # webhook into a mini full sync — ~12 SF OData master-data pulls
    # (companies / departments / locations / positions / personal contact
    # / job-fn / job-family / email-labels) plus ``get_users`` +
    # ``get_jobs``, all before the person upsert + coordinator dispatch.
    # Run inside the request that took tens of seconds against SF's EU
    # datacenter, blowing past the gateway proxy timeout — SF's
    # integration center saw ``504 Gateway Time-out`` and kept replaying.
    #
    # We now authenticate synchronously (a forged / bad-signature request
    # still fails fast with 401 and spawns no work), record the receipt in
    # the audit trail, and hand the actual apply to a background task on a
    # fresh session — returning immediately. The heavy account
    # provisioning (Google mailbox / seats / welcome email) is still
    # deferred to ``hire_date`` by the coordinators' future-hire gate; the
    # scheduled incremental HRIS sync reconciles anything a crashed
    # background task drops. Mirrors ``run_batch_apply_in_background``.
    #
    # The mass-leaver brake stays SYNCHRONOUS (mirrors the batch-confirm
    # path): a forged / corrupted payload that would mass-flip employees
    # to leaver_pending in one POST must surface as an immediate 409, not
    # vanish into the background task. It's a cheap payload scan + one
    # active-person COUNT — no SF round-trips — so it doesn't reintroduce
    # the 504. The background apply re-checks it (belt-and-suspenders).
    brake_service = (
        SuccessFactorsLifecycleService.for_instance(
            db, integration_id=integration_row.id, config=sf_config,
        )
        if integration_row is not None
        else SuccessFactorsLifecycleService(db)
    )
    await brake_service.enforce_mass_leaver_brake(body.events)

    events_received = len(body.events)

    slug = integration_row.slug if integration_row else "primary"
    # Keep stdout suitable for broad operational access: record only bounded
    # routing/body metadata, never the webhook payload (which may contain
    # personal email and phone fields). Starlette caches the body, so hashing
    # it here does not consume it a second time.
    await _log_sf_webhook_receipt(request, slug=slug, body=body)
    # The access-controlled audit row remains the durable recovery source.
    # Cap stored events so a 500-event bootstrap batch cannot bloat the row;
    # normal SF pushes carry one event.
    events_dump = [e.model_dump(mode="json") for e in body.events[:50]]

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_type="system",
        action="hr.successfactors_webhook.receive",
        resource_type="hr_lifecycle_event",
        resource_id=str(events_received),
        status="success",
        detail={
            "events_received": events_received,
            "processing": "async",
            "hris_instance_slug": (
                integration_row.slug if integration_row else None
            ),
            "events": events_dump,
            "events_truncated": events_received > len(events_dump),
        },
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    # ``spawn`` retains a strong reference so the task isn't GC-dropped
    # mid-flight (the failure mode that silently lost every FastAPI
    # BackgroundTask in PR #279). The background task owns its own session
    # — this request's ``db`` is closed once the response returns.
    _spawn_background(
        run_webhook_events_in_background(
            body.events,
            integration_row.id if integration_row else None,
            sf_config if integration_row else None,
        ),
        name=f"hris-webhook-{integration_row.slug if integration_row else 'primary'}",
    )

    payload = {"accepted": True, "events_received": events_received}
    return success_response(data=payload, request=request)


@router.get("/offboarding-cases")
async def list_offboarding_cases(
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    service = SuccessFactorsLifecycleService(db)
    cases = [
        OffboardingCaseResponse(**payload).model_dump(mode="json")
        for payload in await service.list_offboarding_cases()
    ]
    return success_response(data=cases, request=request)


@router.get("/offboarding-cases/{user_id}")
async def get_offboarding_case(
    user_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    service = SuccessFactorsLifecycleService(db)
    payload = OffboardingCaseResponse(**(await service.get_offboarding_case(user_id))).model_dump(mode="json")
    return success_response(data=payload, request=request)


@router.patch("/offboarding-cases/{user_id}/assets/{asset_id}")
async def update_offboarding_asset_status(
    user_id: UUID,
    asset_id: UUID,
    body: OffboardingAssetStatusUpdateRequest,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = SuccessFactorsLifecycleService(db)
    payload = OffboardingAssetResponse(
        **(
            await service.update_offboarding_asset_status(
                user_id=user_id,
                asset_id=asset_id,
                status=body.status,
                note=body.note,
                actor_user_id=current_user.id,
            )
        )
    ).model_dump(mode="json")

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.offboarding_asset.update",
        resource_type="asset_item",
        resource_id=str(asset_id),
        status="success",
        detail=payload,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    return success_response(data=payload, request=request)


@router.patch("/offboarding-cases/{user_id}/software/{software_id}")
async def update_offboarding_software_status(
    user_id: UUID,
    software_id: UUID,
    body: OffboardingAssetStatusUpdateRequest,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = SuccessFactorsLifecycleService(db)
    payload = OffboardingSoftwareResponse(
        **(
            await service.update_offboarding_software_status(
                user_id=user_id,
                software_id=software_id,
                status=body.status,
                note=body.note,
                actor_user_id=current_user.id,
            )
        )
    ).model_dump(mode="json")

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.offboarding_software.update",
        resource_type="asset_software_catalog",
        resource_id=str(software_id),
        status="success",
        detail=payload,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    return success_response(data=payload, request=request)


@router.post("/offboarding-cases/{user_id}/complete")
async def complete_offboarding(
    user_id: UUID,
    body: OffboardingCompletionRequest,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = SuccessFactorsLifecycleService(db)
    payload = OffboardingCaseResponse(
        **(
            await service.complete_offboarding(
                user_id=user_id,
                actor_user_id=current_user.id,
                note=body.note,
            )
        )
    ).model_dump(mode="json")

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.offboarding_case.complete",
        resource_type="identity_person",
        resource_id=str(user_id),
        status="success",
        detail={"note": body.note, "status": payload["status"]},
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    return success_response(data=payload, request=request)


@router.get("/lifecycle-events", summary="List HRIS lifecycle events (admin debug)")
async def list_lifecycle_events(
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=200),
    person_id: UUID | None = Query(
        default=None,
        description="Filter to one person — the immutable ledger for a single employee.",
    ),
    event_type: str | None = Query(
        default=None,
        description="``onboard`` / ``offboard`` (extensible).",
    ),
    trigger_source: str | None = Query(
        default=None,
        description=(
            "``webhook`` (SF push), ``sync_confirm`` (manual confirm), "
            "``initial_seed`` (full bootstrap), ``manual`` (admin override). "
            "Filter ``webhook`` to verify SF event delivery health."
        ),
    ),
    since: datetime | None = Query(
        default=None,
        description="Lower bound on ``created_at`` (UTC).",
    ),
    until: datetime | None = Query(
        default=None,
        description="Upper bound on ``created_at`` (UTC).",
    ),
    person_search: str | None = Query(
        default=None,
        description=(
            "Substring match against display_name / work_email / "
            "name_pinyin / name_pinyin_abbr. P1-J: server-side so a "
            "name on page 3 of the ledger still surfaces."
        ),
        max_length=128,
    ),
):
    """Read the immutable ``hr_lifecycle_events`` ledger.

    Primary debugging surface for the event-driven HRIS rollout: when
    SF's iflow pushes a webhook, an admin can verify the row landed
    here with ``trigger_source='webhook'``; when a sync confirm
    silently no-ops a coordinator (because the webhook already
    handled it), this view shows the prior webhook-tagged row that
    short-circuited the dispatch.

    Admin-only — the JSON ``detail`` blob carries SF-internal codes
    (event/eventReason/emplStatus) plus the canonical employment
    start/end dates, which combined with display_name + email are a
    PII surface that doesn't belong on the read-only HR scope.
    """
    from app.domains.hr.models import HrLifecycleEvent
    from app.identity.person_models import IdentityPerson

    stmt = (
        select(
            HrLifecycleEvent,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
            IdentityPerson.employee_no,
            IdentityPerson.employment_status,
        )
        .join(IdentityPerson, IdentityPerson.id == HrLifecycleEvent.person_id)
    )
    # Build shared filter expression list so the main query + the
    # count query stay in lockstep — the prior code duplicated the
    # filters across both, which created a regression class (one
    # filter added to ``stmt`` but missed in ``count_stmt`` → "total"
    # diverges from "rows shown", a classic ledger-debug footgun).
    from sqlalchemy import func as _sa_func
    filters = []
    if person_id is not None:
        filters.append(HrLifecycleEvent.person_id == person_id)
    if event_type:
        filters.append(HrLifecycleEvent.event_type == event_type)
    if trigger_source:
        filters.append(HrLifecycleEvent.trigger_source == trigger_source)
    if since is not None:
        filters.append(HrLifecycleEvent.created_at >= since)
    if until is not None:
        filters.append(HrLifecycleEvent.created_at <= until)
    person_search_clean = (person_search or "").strip()
    if person_search_clean:
        like = f"%{person_search_clean.lower()}%"
        from sqlalchemy import func as _sa_f2
        from sqlalchemy import or_
        filters.append(or_(
            _sa_f2.lower(IdentityPerson.display_name).like(like),
            _sa_f2.lower(IdentityPerson.work_email).like(like),
            _sa_f2.lower(IdentityPerson.name_pinyin).like(like),
            _sa_f2.lower(IdentityPerson.name_pinyin_abbr).like(like),
        ))

    if filters:
        stmt = stmt.where(*filters)

    # Count BEFORE pagination so the admin UI can render total rows.
    # ``count_stmt`` runs against the same join + filters so total
    # never diverges from rows shown.
    count_stmt = (
        select(_sa_func.count())
        .select_from(HrLifecycleEvent)
        .join(IdentityPerson, IdentityPerson.id == HrLifecycleEvent.person_id)
    )
    if filters:
        count_stmt = count_stmt.where(*filters)
    total = (await db.scalar(count_stmt)) or 0

    stmt = (
        stmt.order_by(HrLifecycleEvent.created_at.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    rows = (await db.execute(stmt)).all()

    items = [
        {
            "id": str(event.id),
            "person_id": str(event.person_id),
            "display_name": display_name,
            "work_email": work_email,
            "display_name_employee_no": employee_no,
            "display_name_employment_status": employment_status,
            "event_type": event.event_type,
            "effective_date": (
                event.effective_date.isoformat() if event.effective_date else None
            ),
            "trigger_source": event.trigger_source,
            "detail": event.detail,
            "created_at": (
                event.created_at.isoformat() if event.created_at else None
            ),
        }
        for event, display_name, work_email, employee_no, employment_status in rows
    ]

    return success_response(
        data=items,
        meta={"page": page, "page_size": page_size, "total": total},
        request=request,
    )


@router.get("/lifecycle-overview")
async def get_lifecycle_overview(
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
    limit: int = Query(default=5, ge=1, le=20),
):
    service = SuccessFactorsLifecycleService(db)
    payload = LifecycleOverviewResponse(**(await service.get_lifecycle_overview(limit=limit))).model_dump(
        mode="json"
    )
    return success_response(data=payload, request=request)


@router.get("/overdue-leavers")
async def list_overdue_leavers(
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    """List leaver_pending users whose termination_date has passed."""
    service = SuccessFactorsLifecycleService(db)
    overdue = await service.list_overdue_leavers()
    return success_response(data=overdue, request=request)


@router.get(
    "/hris-refresh/recent",
    summary="Recent single-person HRIS refreshes — who synced whom, when (HR)",
)
async def hris_refresh_recent(
    request: Request,
    limit: int = Query(20, ge=1, le=100),
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    """Audit trail for the single-person 「立即同步」 button — the last N
    refreshes with actor, target employee, time, and outcome.

    Each refresh is a PAIR of audit rows sharing a trace_id: a
    ``started`` row written synchronously at submit, and a terminal row
    (``success`` / ``blocked`` / ``failure``) written by the background
    job. This endpoint collapses each pair into one item: terminal row
    wins; a lone ``started`` row shows as ``running`` (or
    ``interrupted`` once it is older than 15 minutes — e.g. a deploy
    restarted the backend mid-run). MCP-driven syncs write terminal rows
    only; a terminal row without a started twin is handled the same."""
    from datetime import UTC, datetime, timedelta

    from app.audit.models import AuditLog
    from app.identity.person_models import IdentityPerson

    # Overfetch 2×: pairs collapse into single items.
    raw_rows = (await db.execute(
        select(AuditLog)
        .where(AuditLog.action == "hr.person.hris_refresh")
        .order_by(AuditLog.created_at.desc())
        .limit(limit * 2)
    )).scalars().all()

    # Collapse by trace_id, terminal beats started. Rows come newest-first.
    by_trace: dict[str, AuditLog] = {}
    order: list[str] = []
    for r in raw_rows:
        key = r.trace_id or str(r.id)
        if key not in by_trace:
            by_trace[key] = r
            order.append(key)
        elif by_trace[key].status == "started" and r.status != "started":
            by_trace[key] = r
    stale_cutoff = datetime.now(UTC).replace(tzinfo=None) - timedelta(minutes=15)
    rows = [by_trace[k] for k in order][:limit]

    # Batch-resolve actor + target display names (both are IdentityPerson
    # ids: actor_user_id == the AuthUser/person id; resource_id == the
    # synced person's id).
    person_ids: set[UUID] = set()
    for r in rows:
        if r.actor_user_id:
            person_ids.add(r.actor_user_id)
        if r.resource_id:
            try:
                person_ids.add(UUID(r.resource_id))
            except (ValueError, TypeError):
                pass
    names: dict[UUID, str] = {}
    if person_ids:
        for pid, name in (await db.execute(
            select(IdentityPerson.id, IdentityPerson.display_name)
            .where(IdentityPerson.id.in_(person_ids))
        )).all():
            names[pid] = name

    def _name(pid_str: str | None) -> str | None:
        if not pid_str:
            return None
        try:
            return names.get(UUID(pid_str))
        except (ValueError, TypeError):
            return None

    items = []
    for r in rows:
        detail = r.detail or {}
        # A lone ``started`` row is a live run — or an interrupted one
        # (backend restarted mid-run) once it has aged past the cutoff.
        display_status = r.status
        if r.status == "started":
            display_status = (
                "interrupted"
                if r.created_at and r.created_at < stale_cutoff
                else "running"
            )
        items.append({
            "id": str(r.id),
            "trace_id": r.trace_id,
            "at": r.created_at.isoformat() if r.created_at else None,
            "status": display_status,
            "actor_person_id": str(r.actor_user_id) if r.actor_user_id else None,
            "actor_name": _name(str(r.actor_user_id) if r.actor_user_id else None),
            "person_id": r.resource_id,
            "person_name": _name(r.resource_id),
            "blocked": bool(detail.get("blocked")),
            "block_reason": detail.get("block_reason"),
            "staged_action": detail.get("staged_action"),
            "helix_applied": bool(detail.get("helix_applied")),
            "changed_fields": detail.get("changed_fields") or [],
            "feishu_status": detail.get("feishu_status"),
        })
    return success_response(data={"items": items}, request=request)


@router.post(
    "/people/{person_id}/hris-refresh",
    response_model=None,
    summary="Sync ONE person from SuccessFactors → Helix → Feishu now (HR)",
)
async def hris_refresh_person(
    person_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """HR "sync this person now" — submit + poll, NOT synchronous.

    A single-person refresh is SF pull (incl. master-data catalogs) →
    staged-batch confirm → synchronous Feishu push. On the CN instance
    that's ~30s; on the EU instance it was measured at ~3 MINUTES
    (2026-07-23, Chi Yeung Yip P10018). A synchronous HTTP request that
    long is structurally broken: the browser/axios timeout aborts, uvicorn
    cancels the handler coroutine mid-flight (``CancelledError`` bypasses
    ``except Exception``), and the run finishes server-side with NO audit
    row — the user sees nothing in the UI and nothing in the history.
    Exactly the incident this endpoint's third fix addresses.

    So this handler mirrors the platform's confirm-batch pattern (apply
    runs off the request thread, UI polls to terminal):

    1. Validate cheaply (person exists, is SF-sourced).
    2. Write a ``status='started'`` audit row (committed immediately —
       the attempt is ALWAYS visible in the history, whatever happens).
    3. ``spawn`` the coordinator in the background — client disconnects
       can no longer cancel it.
    4. Return the trace_id; the UI polls ``GET /hris-refresh/status/{trace_id}``
       until the background job writes the terminal row (same trace_id).

    The lifecycle-transition gate (new hire / rehire / termination →
    blocked) lives in ``hris_person_refresh_coordinator`` unchanged.
    """
    from app.core.background import spawn
    from app.identity.person_models import IdentityPerson

    person = await db.get(IdentityPerson, person_id)
    if person is None:
        raise NotFoundError("IdentityPerson", str(person_id))
    if person.source_system != "successfactors" or not person.hris_person_id:
        raise AppException(
            code="HR_PERSON_NOT_HRIS_SOURCED",
            message="该人员不是来自 SuccessFactors 的员工记录,无法从 HRIS 单人同步。",
            status_code=400,
        )

    trace_id = request.state.trace_id
    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.person.hris_refresh",
        resource_type="identity_person",
        resource_id=str(person_id),
        status="started",
        detail={"phase": "started"},
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    spawn(
        _run_hris_refresh_bg(
            person_id=person_id,
            actor_user_id=current_user.id,
            trace_id=trace_id,
        ),
        name=f"hris_refresh:{person_id}",
    )

    return success_response(
        data={"started": True, "trace_id": trace_id}, request=request,
    )


async def _run_hris_refresh_bg(
    *,
    person_id: UUID,
    actor_user_id: str,
    trace_id: str,
) -> None:
    """Background body of the single-person refresh.

    Runs the coordinator on its own session and ALWAYS lands a terminal
    audit row (success / blocked / failure) with the same ``trace_id`` as
    the ``started`` row — the polling endpoint and the history table key
    off that pairing. The full masked result dict is stored in
    ``detail.result`` so the UI can render the same step-by-step outcome
    the old synchronous response carried.
    """
    from app.core.coordinators.hris_person_refresh_coordinator import (
        refresh_person_from_hris,
    )
    from app.core.database import async_session_factory

    status = "failure"
    detail: dict = {}
    try:
        async with async_session_factory() as db:
            result = await refresh_person_from_hris(
                db,
                person_id=person_id,
                actor_user_id=actor_user_id,
            )
        status = "blocked" if result.blocked else "success"
        detail = {
            "hris_user_id": result.hris_user_id,
            "staged_action": result.staged_action,
            "blocked": result.blocked,
            "block_reason": result.block_reason,
            "helix_applied": result.helix_applied,
            "changed_fields": [c["field"] for c in result.changed_fields],
            "feishu_status": (
                result.feishu_push.status if result.feishu_push else None
            ),
            # Full (already-masked) result for the UI's step renderer.
            "result": result.to_dict(),
        }
    except Exception as exc:  # noqa: BLE001
        logger.exception(
            "hris_refresh bg: coordinator failed for person=%s", person_id,
        )
        detail = {
            "error": "HRIS_REFRESH_FAILED",
            "error_type": type(exc).__name__,
        }
    finally:
        # Terminal row on a FRESH session — the coordinator session may be
        # poisoned after a DB error, and this row is what flips the UI's
        # "running" into a real outcome. If even this write fails the
        # started row stays lone and the UI ages it into "interrupted".
        try:
            async with async_session_factory() as audit_session:
                await AuditService(audit_session).log(
                    trace_id=trace_id,
                    actor_user_id=actor_user_id,
                    actor_type="user",
                    action="hr.person.hris_refresh",
                    resource_type="identity_person",
                    resource_id=str(person_id),
                    status=status,
                    detail=detail,
                )
                await audit_session.commit()
        except Exception:  # noqa: BLE001
            logger.exception(
                "hris_refresh bg: failed to write terminal audit for "
                "person=%s trace=%s", person_id, trace_id,
            )


@router.get(
    "/hris-refresh/status/{trace_id}",
    summary="Poll a single-person HRIS refresh to terminal (HR)",
)
async def hris_refresh_status(
    trace_id: str,
    request: Request,
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    """Terminal-or-running state for one refresh, keyed by trace_id.

    ``done=false`` while only the ``started`` row exists; once the
    background job lands the terminal row this returns its status and the
    stored result dict (same shape the old synchronous response had)."""
    from app.audit.models import AuditLog

    rows = (await db.execute(
        select(AuditLog)
        .where(
            AuditLog.action == "hr.person.hris_refresh",
            AuditLog.trace_id == trace_id,
        )
        .order_by(AuditLog.created_at.asc())
    )).scalars().all()
    if not rows:
        raise NotFoundError("hris_refresh", trace_id)

    terminal = next(
        (r for r in rows if r.status in ("success", "blocked", "failure")),
        None,
    )
    if terminal is None:
        return success_response(
            data={"done": False, "status": "running"}, request=request,
        )
    detail = terminal.detail or {}
    return success_response(
        data={
            "done": True,
            "status": terminal.status,
            "result": detail.get("result"),
            "error": detail.get("error"),
        },
        request=request,
    )


@router.post(
    "/people/{person_id}/mark-leaver",
    response_model=None,
    summary="Manually flip an active person to leaver_pending and start offboarding",
)
async def mark_person_leaver_manual(
    person_id: UUID,
    body: ManualMarkLeaverRequest,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Admin-only manual offboarding entry point.

    Mirrors the HRIS-driven path: sets ``employment_status='leaver_pending'``
    + ``termination_date`` on the IdentityPerson row, then calls
    ``OffboardingService.start_process`` which creates the
    OffboardingProcess, scans blockers, and pushes the force-logout.
    Idempotent — re-calling on a person already in ``leaver_pending``
    returns 409 to prevent silent date overwrites.
    """
    from app.domains.offboarding.service import OffboardingService
    from app.identity.person_models import IdentityPerson

    person = await db.get(IdentityPerson, person_id)
    if person is None or person.deleted_at is not None:
        raise AppException(
            code="PERSON_NOT_FOUND",
            message=f"Person {person_id} not found",
            status_code=404,
        )
    if person.employment_status != "active":
        raise AppException(
            code="PERSON_NOT_ACTIVE",
            message=(
                f"Person {person.work_email} is already in "
                f"'{person.employment_status}' status — cannot re-mark "
                "as leaver. Cancel the existing offboarding process first."
            ),
            status_code=409,
        )

    # P1-5 fix (2026-05-08): bound the leaver date so a typo doesn't
    # tank the flow. Range is intentionally generous (back 7 days for
    # backdated paper exits HR fills in a few days late, forward 365
    # days for long-notice scheduled exits) — outside that window the
    # admin almost certainly typed wrong, and the downstream cron
    # ``hr.overdue_leavers_check`` would treat a far-past date as a
    # zero-grace immediate transition to ``left``.
    today = date.today()
    earliest = today - timedelta(days=7)
    latest = today + timedelta(days=365)
    if body.last_working_day < earliest:
        raise AppException(
            code="LAST_WORKING_DAY_TOO_OLD",
            message=(
                f"last_working_day {body.last_working_day.isoformat()} is "
                f"more than 7 days in the past. If this is correct, the "
                f"employee should already be in 'left' status — fix the "
                f"current state first or use the HRIS sync path. "
                f"Allowed range: {earliest.isoformat()} → "
                f"{latest.isoformat()}."
            ),
            status_code=400,
        )
    if body.last_working_day > latest:
        raise AppException(
            code="LAST_WORKING_DAY_TOO_FAR",
            message=(
                f"last_working_day {body.last_working_day.isoformat()} is "
                f"more than 365 days in the future. Long-notice exits "
                f"should land via HRIS where the date is canonical. "
                f"Allowed range: {earliest.isoformat()} → "
                f"{latest.isoformat()}."
            ),
            status_code=400,
        )

    person.employment_status = "leaver_pending"
    person.termination_date = body.last_working_day
    await db.flush()

    off_svc = OffboardingService(db)
    proc = await off_svc.start_process(
        person.id,
        body.last_working_day,
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
    )

    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.person.mark_leaver_manual",
        resource_type="identity_person",
        resource_id=str(person.id),
        status="success",
        detail={
            "work_email": person.work_email,
            "last_working_day": body.last_working_day.isoformat(),
            "reason": body.reason,
            "offboarding_process_id": str(proc.id),
        },
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )

    await db.commit()

    # D-2 fix (2026-05-08): post-commit side effects. Force-logout +
    # leaver/manager notifications fire AFTER the offboarding process
    # is durably committed, so a transaction rollback can never leave
    # "Google session killed but Helix has no record" or
    # "Manager Feishu'd but the process row never wrote" inconsistencies.
    # ``dispatch_post_commit_actions`` itself swallows internal failures.
    # We wrap the second commit too — by this point the offboarding is
    # durable, so even a deadlock on the audit-only second commit must
    # not bubble out as a 500: the operator sees "marked successfully"
    # in the response, the post_commit dict surfaces what fell through.
    _hr_logger = get_safe_logger("app.domains.hr.router")
    post_commit_summary: dict = {}
    try:
        post_commit_summary = await off_svc.dispatch_post_commit_actions(
            person.id, proc.id,
            trace_id=request.state.trace_id,
            actor_user_id=current_user.id,
        )
        await db.commit()
    except Exception as exc:  # noqa: BLE001
        _hr_logger.exception(
            "mark_leaver post-commit dispatch second-commit failed for "
            "person %s — process is committed; force-logout / "
            "notifications may be partial. Operator can manually "
            "re-trigger from the offboarding panel. error_type=%s",
            person.id,
            type(exc).__name__,
        )
        # The dispatch exception may contain provider responses, signed URLs,
        # or SQL parameters. Keep those out of the admin response; trace_id
        # plus the safe log's exception type retain correlation.
        post_commit_summary = {
            "post_commit_commit_error": "INTERNAL_ERROR",
            "google_force_logout_pushed": False,
            "notifications": {"leaver": None, "manager": None},
        }
        # Best-effort rollback of any uncommitted dispatch state so the
        # session is clean for the response serialization that follows.
        try:
            await db.rollback()
        except Exception:  # noqa: BLE001
            pass

    # Pre-launch P2 fix (Round 1, CR-015): post_commit is a typed
    # field on the response now (``ManualMarkLeaverPostCommitResult``)
    # instead of a free-form dict merged after ``model_dump``.
    from app.domains.hr.schemas import (  # noqa: PLC0415
        ManualMarkLeaverPostCommitResult,
    )
    try:
        post_commit_typed = ManualMarkLeaverPostCommitResult(
            **(post_commit_summary or {})
        )
    except Exception:  # noqa: BLE001
        # If the dispatcher returned an unexpected shape, fall back to
        # an empty result rather than 500-ing the whole response. The
        # process is already durable.
        post_commit_typed = ManualMarkLeaverPostCommitResult()
    payload = ManualMarkLeaverResponse(
        person_id=str(person.id),
        employment_status=person.employment_status,
        termination_date=person.termination_date,
        offboarding_process_id=str(proc.id),
        post_commit=post_commit_typed,
    ).model_dump(mode="json")
    return success_response(data=payload, request=request)


@router.post("/overdue-leavers/transition")
async def transition_overdue_leavers(
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Batch-transition overdue leaver_pending users to 'left' status."""
    service = SuccessFactorsLifecycleService(db)
    summary = await service.transition_overdue_leavers()

    if summary["transitioned_count"] > 0:
        audit = AuditService(db)
        await audit.log(
            trace_id=request.state.trace_id,
            actor_user_id=current_user.id,
            actor_type="user",
            action="hr.overdue_leavers.transition",
            resource_type="identity_person",
            resource_id=f"batch:{summary['transitioned_count']}",
            status="success",
            detail=summary,
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("user-agent"),
        )

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


# ── Leave / time-off (SF EC Time-Off mirror) ─────────────────────────
#
# Self / manager / HR-admin scoped, mirroring the assets & accounts
# visibility model: an employee sees their own balances + ledger, a
# manager sees their (transitive) reports, and HR/global_admin/auditor/
# management see everyone (``can_view_hr_data``). The data is read-only as a
# business record because SuccessFactors owns it; refresh endpoints still
# mutate Helix's mirror and therefore use the narrower write-side gate below.


def _serialize_posting(p) -> LeavePostingResponse:
    amount = float(p.booking_amount) if p.booking_amount is not None else 0.0
    bal_after = float(p.balance_after) if p.balance_after is not None else None
    return LeavePostingResponse(
        id=p.id,
        booking_type=p.booking_type,
        booking_amount=amount,
        booking_unit=p.booking_unit,
        booking_date=p.booking_date,
        # 之前的剩余假期 = 之后 − 本行数额 (SF shows both columns).
        balance_before=(None if bal_after is None else round(bal_after - amount, 2)),
        balance_after=bal_after,
        created_by=p.created_by_source,
        created_at_source=p.created_at_source,
        comment=p.comment,
        reason=p.reason,
        leave_type=p.leave_type,
        request_status=p.request_status,
    )


class _LeaveViewContext(NamedTuple):
    """Per-request context for rendering leave cards.

    Loaded once per request rather than per account: the type-override setting
    is a single platform-settings read, and every card in one response must be
    counted against the SAME "today" or two cards can disagree by a day.
    """

    overrides: dict[str, str]
    today: date


async def _leave_view_context(db: AsyncSession) -> _LeaveViewContext:
    from app.domains.hr.leave_insights_service import load_type_overrides

    overrides = await load_type_overrides(db)
    # Anchored to the company operating timezone, exactly like the admin
    # 休假健康洞察 board — the employee card and the admin board must never
    # quote a different number of days remaining for the same account.
    today = datetime.now(UTC).astimezone(ZoneInfo("Asia/Shanghai")).date()
    return _LeaveViewContext(overrides=overrides, today=today)


def _leave_forfeit_signals(
    a, ctx: _LeaveViewContext | None,
) -> tuple[bool, int | None]:
    """``(forfeits, expires_in_days)`` for one leave account.

    "Use it or lose it" is a property of the account TYPE (法定年假 forfeits;
    调休 banks; 病假 is not a balance you spend down), and which vendor code
    means what is tenant-configurable — so the classification is read from the
    leave domain's own ``forfeits_at_period_end`` (the SAME function and the
    same tenant overrides the admin 休假健康洞察 board's 临期未休 total uses)
    rather than re-guessed here or, worse, in the frontend.
    Employees previously had no way at all to see that a balance was about to
    lapse: 临期未休 existed only on the admin insights page.
    """
    from app.domains.hr.leave_insights_service import forfeits_at_period_end

    if ctx is None or a.account_closed:
        return False, None
    forfeits = forfeits_at_period_end(a.account_type, ctx.overrides)
    deadline = a.booking_end_date or a.end_date
    if deadline is None:
        return forfeits, None
    remaining = (deadline - ctx.today).days
    if remaining < 0:
        # Past the deadline there is nothing to count down to. This used to
        # clamp to 0 — which the card read as "expires TODAY" and rendered in
        # the most urgent red state, forever, for any past-year bucket SF
        # never formally closed (``list_accounts`` returns those by default).
        return forfeits, None
    return forfeits, remaining


def _serialize_account(
    a, *, postings=None, ctx: _LeaveViewContext | None = None,
) -> LeaveAccountResponse:
    forfeits, expires_in_days = _leave_forfeit_signals(a, ctx)
    return LeaveAccountResponse(
        id=a.id,
        account_type=a.account_type,
        account_type_label=a.account_type_label,
        fiscal_year=a.fiscal_year,
        start_date=a.start_date,
        end_date=a.end_date,
        booking_end_date=a.booking_end_date,
        account_closed=a.account_closed,
        is_current=a.is_current,
        forfeits=forfeits,
        expires_in_days=expires_in_days,
        balance_days=float(a.balance_days) if a.balance_days is not None else None,
        entitled_days=float(a.entitled_days) if a.entitled_days is not None else None,
        used_days=float(a.used_days) if a.used_days is not None else None,
        postings=(
            [_serialize_posting(p) for p in postings]
            if postings is not None
            else None
        ),
    )


async def _assert_can_view_person_leave(
    db: AsyncSession, current_user: AuthUser, person_id: str,
) -> None:
    """Self / manager-of / HR-admin gate for a person's leave data."""
    if str(current_user.id) == str(person_id):
        return
    if can_view_hr_data(current_user.role_names):
        return
    if current_user.is_manager:
        subs = await subordinate_person_ids(
            db, current_user.id, scope=SubordinateScope.ALL,
        )
        if any(str(s) == str(person_id) for s in subs):
            return
    raise ForbiddenError(
        "You can only view your own or your subordinates' leave balances",
    )


async def _assert_can_refresh_person_leave(
    db: AsyncSession, current_user: AuthUser, person_id: str,
) -> None:
    """Write-side leave refresh gate.

    Self, managers of the target, HR and global admins may refresh. Auditor
    and management remain read-only even though they can view the same leave
    ledger; reusing the read helper here would let governance roles trigger
    vendor calls and local writes.
    """
    if str(current_user.id) == str(person_id):
        return
    if has_role(current_user.role_names, GLOBAL_ADMIN_ROLE, HR_ROLE):
        return
    if current_user.is_manager:
        subs = await subordinate_person_ids(
            db, current_user.id, scope=SubordinateScope.ALL,
        )
        if any(str(s) == str(person_id) for s in subs):
            return
    raise ForbiddenError(
        "You can only refresh your own or your subordinates' leave balances",
    )


@router.get("/leave/me", summary="My leave balances + accounts (self-service)")
async def get_my_leave(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    accounts = await HrLeaveService(db).list_accounts(str(current_user.id))
    ctx = await _leave_view_context(db)
    data = {
        "person_id": str(current_user.id),
        "accounts": [_serialize_account(a, ctx=ctx).model_dump() for a in accounts],
    }
    return success_response(data=data, request=request)


@router.get(
    "/leave/persons/{person_id}",
    summary="A person's leave balances (manager / HR scope)",
)
async def get_person_leave(
    person_id: str,
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    await _assert_can_view_person_leave(db, current_user, person_id)
    accounts = await HrLeaveService(db).list_accounts(person_id)
    ctx = await _leave_view_context(db)
    data = {
        "person_id": person_id,
        "accounts": [_serialize_account(a, ctx=ctx).model_dump() for a in accounts],
    }
    return success_response(data=data, request=request)


@router.get(
    "/leave/accounts/{account_id}",
    summary="One leave account + its full ledger (账户假期记录)",
)
async def get_leave_account(
    account_id: str,
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    svc = HrLeaveService(db)
    account = await svc.get_account(account_id)
    if account is None or account.deleted_at is not None:
        raise NotFoundError("Leave account not found")
    await _assert_can_view_person_leave(db, current_user, account.person_id)
    postings = await svc.list_postings(account_id)
    ctx = await _leave_view_context(db)
    data = _serialize_account(account, postings=postings, ctx=ctx).model_dump()
    return success_response(data=data, request=request)


@router.post(
    "/leave/sync",
    summary="Refresh leave balances + records from SuccessFactors (admin)",
)
async def trigger_leave_sync(
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Fan-out a fresh SF EC Time-Off pull across every enabled HRIS
    instance, in the background. Returns ``202``-style immediately; the
    daily ``hr.leave_account_sync`` cron is the steady-state path."""
    await AuditService(db).log(
        trace_id=request.state.trace_id,
        actor_user_id=current_user.id,
        actor_type="user",
        action="hr.leave.manual_sync",
        resource_type="hr_leave",
        resource_id="all",
        status="success",
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    _spawn_background(run_leave_sync_in_background())
    return success_response(
        data={"status": "started"}, request=request,
    )


async def _refresh_one_person_leave(
    db: AsyncSession,
    request: Request,
    current_user: AuthUser,
    person_id: str,
) -> dict:
    """Shared body for the per-person / self leave "refresh from SF now"
    flows. Re-pulls ONE person's SF Time-Off SYNCHRONOUSLY (unlike the
    admin all-roster ``/leave/sync`` which is background), then returns the
    freshly-synced accounts so the caller renders current data in one
    round-trip. Caller must have already passed the view-scope gate."""
    from app.identity.person_models import IdentityPerson

    person = await db.get(IdentityPerson, person_id)
    if person is None or person.deleted_at is not None:
        raise NotFoundError("Person not found")
    async def _audit_before_commit(summary: dict) -> None:
        await AuditService(db).log(
            trace_id=request.state.trace_id,
            actor_user_id=current_user.id,
            actor_type="user",
            action="hr.leave.manual_sync_person",
            resource_type="identity_person",
            resource_id=str(person_id),
            # A pull where every HRIS instance failed is not a successful
            # sync, and an audit trail that says otherwise is worse than no
            # audit trail.
            status=(
                "partial" if summary.get("failed_instances") else "success"
            ),
            detail=summary,
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("user-agent"),
        )

    summary = await sync_one_person_leave_now(
        db,
        person,
        actor_id=current_user.id,
        before_commit=_audit_before_commit,
    )
    accounts = await HrLeaveService(db).list_accounts(person_id)
    ctx = await _leave_view_context(db)
    return {
        "person_id": person_id,
        "accounts": [_serialize_account(a, ctx=ctx).model_dump() for a in accounts],
        # `synced` is true only when a pull actually happened AND every
        # instance answered. `skipped` covers the non-SF person; the
        # failure list covers the outage.
        "synced": (
            summary.get("skipped") is None and not summary.get("failed_instances")
        ),
        "summary": summary,
    }


@router.post(
    "/leave/me/sync",
    summary="Refresh MY leave balances from SuccessFactors now (self-service)",
)
async def sync_my_leave(
    request: Request,
    current_user: AuthUser = Depends(require_active_employee),
    db: AsyncSession = Depends(get_db),
):
    """Synchronous SF re-pull of the caller's own Time-Off, returning the
    fresh accounts. Lets an employee pull a just-edited leave request
    without waiting for the daily cron."""
    data = await _refresh_one_person_leave(
        db, request, current_user, str(current_user.id),
    )
    return success_response(data=data, request=request)


@router.post(
    "/leave/persons/{person_id}/sync",
    summary="Refresh ONE person's leave from SuccessFactors now (manager / HR)",
)
async def sync_person_leave(
    person_id: str,
    request: Request,
    current_user: AuthUser = Depends(require_active_employee),
    db: AsyncSession = Depends(get_db),
):
    """Synchronous SF re-pull of a single person's Time-Off (self /
    manager-of / HR-admin), returning the fresh accounts. The targeted
    counterpart to the admin all-roster ``/leave/sync`` — for "this one
    employee just changed their leave request" rather than a nightly
    full refresh."""
    await _assert_can_refresh_person_leave(db, current_user, person_id)
    data = await _refresh_one_person_leave(
        db, request, current_user, person_id,
    )
    return success_response(data=data, request=request)


_LEAVE_BOOKING_TYPE_LABEL = {
    "ACCRUAL": "累计假期",
    "EMPLOYEE_TIME": "休假预订",
    "DEDUCTION": "扣减",
    "MANUAL_ADJUSTMENT": "手动调整",
}


@router.get(
    "/leave/persons/{person_id}/export.xlsx",
    summary="导出某人的假期余额与记录 (xlsx)",
)
async def export_person_leave(
    person_id: str,
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Flatten a person's leave accounts + ledger into one xlsx sheet.
    Self / manager / HR scope (same gate as the read endpoints). Goes
    through the platform's single ``xlsx_export_response`` primitive."""
    from app.core.exporting import xlsx_export_response

    await _assert_can_view_person_leave(db, current_user, person_id)
    svc = HrLeaveService(db)
    accounts = await svc.list_accounts(person_id)
    rows: list[dict] = []
    for a in accounts:
        postings = await svc.list_postings(str(a.id))
        for p in postings:
            amt = float(p.booking_amount) if p.booking_amount is not None else 0.0
            bal_after = float(p.balance_after) if p.balance_after is not None else None
            rows.append({
                "account": a.account_type_label or a.account_type,
                "fiscal_year": a.fiscal_year,
                "booking_date": p.booking_date.isoformat() if p.booking_date else "",
                "booking_type": _LEAVE_BOOKING_TYPE_LABEL.get(
                    p.booking_type, p.booking_type,
                ),
                "amount": amt,
                "balance_before": (
                    None if bal_after is None else round(bal_after - amt, 2)
                ),
                "balance_after": bal_after,
                "reason": p.reason or "",
                "status": p.request_status or "",
                "created_by": p.created_by_source or "",
            })
    columns = [
        ("account", "假期类型"),
        ("fiscal_year", "年度"),
        ("booking_date", "发布日期"),
        ("booking_type", "类型"),
        ("amount", "数额(天)"),
        ("balance_before", "之前的剩余假期"),
        ("balance_after", "之后的剩余假期"),
        ("reason", "请假事由"),
        ("status", "审批"),
        ("created_by", "创建者"),
    ]
    return await xlsx_export_response(
        db, request, current_user,
        columns=columns, rows=rows, total_matched=len(rows),
        filename_stem=f"leave_{person_id}",
        sheet_title="假期记录",
        audit_action="hr.leave.export",
        audit_resource_type="hr_leave",
    )


# ── Leave-health insights (HR / management read-only governance view) ─────
# Counter-intuitive lens: low annual-leave utilisation by a BU / team / lead
# is a team-health risk signal, not a badge of diligence. Read-only; admits
# global_admin / hr / auditor / management via require_hr_read(). No pushes.

_INSIGHT_DIMENSIONS = (DIMENSION_BU, DIMENSION_DEPARTMENT, DIMENSION_LEAD)
# Stable health sort weight: most-concerning first when ranking.
def _validate_dimension(dimension: str) -> str:
    if dimension not in _INSIGHT_DIMENSIONS:
        raise AppValidationError(
            f"dimension must be one of {', '.join(_INSIGHT_DIMENSIONS)}",
        )
    return dimension


def _sorted_groups(
    rows: list, *, eligible_only: bool, include_uncovered: bool = False,
) -> list:
    """Rank GroupInsight rows: rankable groups (covered ≥ guard) by ascending
    utilisation (lowest = most concerning) first, then small-but-has-data
    (``insufficient_data``) groups by member count, and finally — only when
    asked — the ``uncovered`` groups (nobody has a leave account; mostly
    non-CN teams) which are excluded by default so they never read as a
    low-usage problem. ``eligible_only`` keeps just the rankable groups."""
    eligible = [
        r for r in rows
        if r.health not in (HEALTH_INSUFFICIENT, HEALTH_UNCOVERED)
    ]
    insufficient = [r for r in rows if r.health == HEALTH_INSUFFICIENT]
    uncovered = [r for r in rows if r.health == HEALTH_UNCOVERED]
    eligible.sort(
        key=lambda r: (
            r.utilisation if r.utilisation is not None else 1e9,
            -r.covered_count,
        )
    )
    if eligible_only:
        return eligible
    insufficient.sort(key=lambda r: -r.member_count)
    out = eligible + insufficient
    if include_uncovered:
        uncovered.sort(key=lambda r: -r.member_count)
        out += uncovered
    return out


@router.get(
    "/leave/insights/overview",
    summary="休假健康洞察 · 总览（公司基线 + 需关注 Top）",
)
async def leave_insights_overview(
    request: Request,
    dimension: str = Query(DIMENSION_LEAD),
    limit: int = Query(5, ge=1, le=20),
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    dimension = _validate_dimension(dimension)
    svc = LeaveInsightsService(db)
    ctx = await svc.build_context()
    groups = svc.group(dimension, ctx)

    active = len(ctx.metrics)
    covered = sum(1 for m in ctx.metrics if m.has_quota)
    health_dist = {
        HEALTH_HEALTHY: 0, HEALTH_NORMAL: 0,
        HEALTH_ATTENTION: 0, HEALTH_INSUFFICIENT: 0, HEALTH_UNCOVERED: 0,
    }
    for g in groups:
        health_dist[g.health] = health_dist.get(g.health, 0) + 1

    baseline = CompanyBaseline(
        fiscal_year=ctx.fiscal_year,
        elapsed_fraction=ctx.elapsed_fraction,
        as_of=ctx.as_of.isoformat(),
        active_population=active,
        covered_population=covered,
        coverage_pct=(round(covered / active, 4) if active else None),
        annual_entitled_total=ctx.company_entitled_total,
        annual_used_total=ctx.company_used_total,
        utilisation=ctx.company_utilisation,
        pace_index=ctx.company_pace_index,
        forfeit_at_risk_total=round(
            sum(m.forfeit_at_risk for m in ctx.metrics), 2,
        ),
        comp_banked_total=round(sum(m.comp_balance for m in ctx.metrics), 2),
        health_distribution=health_dist,
    )
    attention = [
        GroupInsightRow(**g.as_dict())
        for g in _sorted_groups(groups, eligible_only=True)[:limit]
    ]
    payload = OverviewResponse(
        baseline=baseline, thresholds=ctx.thresholds, attention_groups=attention,
    )
    return success_response(
        data=payload.model_dump(mode="json"),
        request=request,
        meta={"dimension": dimension},
    )


@router.get(
    "/leave/insights/groups",
    summary="休假健康洞察 · 分维度排名（BU / 团队 / Lead）",
)
async def leave_insights_groups(
    request: Request,
    dimension: str = Query(DIMENSION_BU),
    include_insufficient: bool = Query(True),
    include_uncovered: bool = Query(False),
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    dimension = _validate_dimension(dimension)
    svc = LeaveInsightsService(db)
    ctx = await svc.build_context()
    groups = svc.group(dimension, ctx)
    # Uncovered groups (nobody has a leave account — mostly non-CN teams) are
    # excluded from the ranking by default so they don't read as a low-usage
    # problem; the count is surfaced so the UI can say "+N uncovered".
    uncovered_count = sum(1 for g in groups if g.health == HEALTH_UNCOVERED)
    ordered = _sorted_groups(
        groups,
        eligible_only=not include_insufficient,
        include_uncovered=include_uncovered,
    )
    rows = [GroupInsightRow(**g.as_dict()).model_dump(mode="json") for g in ordered]
    return success_response(
        data={
            "dimension": dimension,
            "company_utilisation": ctx.company_utilisation,
            "company_pace_index": ctx.company_pace_index,
            "fiscal_year": ctx.fiscal_year,
            "as_of": ctx.as_of.isoformat(),
            "uncovered_count": uncovered_count,
            "groups": rows,
        },
        request=request,
        meta={"dimension": dimension, "row_count": len(rows)},
    )


@router.get(
    "/leave/insights/members",
    summary="休假健康洞察 · 某组/团队的成员明细（下钻）",
)
async def leave_insights_members(
    request: Request,
    dimension: str = Query(...),
    key: str = Query(...),
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    dimension = _validate_dimension(dimension)
    svc = LeaveInsightsService(db)
    ctx = await svc.build_context()
    members = svc.members_of(dimension, key, ctx)
    if not members:
        raise NotFoundError("No members for this group")
    # Small-group privacy guard: a group below min_group_size covered members
    # is `insufficient_data` in the ranking precisely so an individual is never
    # singled out under a "team" label — the drill-down must honour the SAME
    # guard, otherwise a caller (esp. management / auditor) could reconstruct
    # one person's exact leave usage by drilling into a 1-2 person team.
    if svc.below_privacy_guard(members, ctx):
        raise AppValidationError(
            "This group is too small to show individual detail "
            "(small-group privacy guard).",
        )
    # Lowest utilisation first; no-quota members last.
    members_sorted = sorted(
        members,
        key=lambda m: (
            0 if m.has_quota else 1,
            m.utilisation if m.utilisation is not None else 1e9,
        ),
    )
    label = key
    if dimension == DIMENSION_LEAD:
        lead = ctx.by_person.get(key)
        if lead is not None:
            label = lead.display_name
    payload = GroupMembersResponse(
        dimension=dimension,
        key=key,
        label=label,
        company_utilisation=ctx.company_utilisation,
        members=[
            MemberRow(
                person_id=m.person_id,
                display_name=m.display_name,
                employee_no=m.employee_no,
                department=m.department,
                bu=m.bu,
                annual_entitled=round(m.annual_entitled, 2),
                annual_used=round(m.annual_used, 2),
                annual_balance=round(m.annual_balance, 2),
                forfeit_at_risk=round(m.forfeit_at_risk, 2),
                comp_balance=round(m.comp_balance, 2),
                utilisation=m.utilisation,
                has_quota=m.has_quota,
            )
            for m in members_sorted
        ],
    )
    return success_response(data=payload.model_dump(mode="json"), request=request)


@router.get(
    "/leave/insights/trend",
    summary="休假健康洞察 · 趋势（自上线起逐日累积）",
)
async def leave_insights_trend(
    request: Request,
    dimension: str = Query(...),
    key: str = Query(...),
    days: int = Query(180, ge=7, le=730),
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):

    # Trend additionally allows the company baseline series (so a group line
    # can be drawn against company over time); the per-group dimensions still
    # validate to bu/department/lead.
    if dimension == DIMENSION_COMPANY:
        if key != COMPANY_KEY:
            raise AppValidationError("company trend key must be the baseline key")
    else:
        dimension = _validate_dimension(dimension)
        # Honour the same small-group privacy guard as the ranking + members:
        # never expose a sub-min group's series (its utilisation == one or two
        # individuals'). Return an empty series rather than leaking history.
        svc = LeaveInsightsService(db)
        ctx = await svc.build_context()
        members = svc.members_of(dimension, key, ctx)
        if not members or svc.below_privacy_guard(members, ctx):
            return success_response(
                data=TrendResponse(dimension=dimension, key=key, points=[]).model_dump(mode="json"),
                request=request,
            )
    cutoff = (datetime.now(UTC).date() - timedelta(days=days))
    rows = (
        await db.execute(
            select(HrLeaveInsightSnapshot)
            .where(
                HrLeaveInsightSnapshot.dimension == dimension,
                HrLeaveInsightSnapshot.group_key == key,
                HrLeaveInsightSnapshot.snapshot_date >= cutoff,
            )
            .order_by(HrLeaveInsightSnapshot.snapshot_date)
        )
    ).scalars().all()
    points = [
        TrendPoint(
            snapshot_date=r.snapshot_date.isoformat(),
            utilisation=r.utilisation,
            pace_index=r.pace_index,
            vs_company=r.vs_company,
            covered_count=r.covered_count,
            health=r.health,
        )
        for r in rows
    ]
    payload = TrendResponse(dimension=dimension, key=key, points=points)
    return success_response(data=payload.model_dump(mode="json"), request=request)


@router.get(
    "/leave/insights/export.xlsx",
    summary="休假健康洞察 · 导出分维度排名 (xlsx)",
)
async def leave_insights_export(
    request: Request,
    dimension: str = Query(DIMENSION_BU),
    current_user: AuthUser = Depends(require_hr_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.core.exporting import xlsx_export_response

    dimension = _validate_dimension(dimension)
    svc = LeaveInsightsService(db)
    ctx = await svc.build_context()
    ordered = _sorted_groups(svc.group(dimension, ctx), eligible_only=False)
    health_label = {
        HEALTH_HEALTHY: "健康",
        HEALTH_NORMAL: "一般",
        HEALTH_ATTENTION: "需关注",
        HEALTH_INSUFFICIENT: "数据不足",
        HEALTH_UNCOVERED: "未覆盖",
    }
    rows = []
    for g in ordered:
        # Redact metrics for sub-min groups (same small-group privacy guard as
        # the ranking / members / trend) so the export can't be used to
        # reconstruct a tiny team's — or an individual's — leave numbers.
        d = g.as_dict(redact=True)
        pct100 = lambda v: None if v is None else round(v * 100, 1)  # noqa: E731
        days1 = lambda v: None if v is None else round(v, 1)  # noqa: E731
        rows.append({
            "label": d["label"],
            "members": d["member_count"],
            "covered": d["covered_count"],
            "utilisation": pct100(d["utilisation"]),
            "vs_company": pct100(d["vs_company"]),
            "low_utilisers": d["low_utiliser_count"],
            "entitled": days1(d["annual_entitled_total"]),
            "used": days1(d["annual_used_total"]),
            "forfeit": days1(d["forfeit_at_risk_total"]),
            "comp": days1(d["comp_banked_total"]),
            "health": health_label.get(g.health, g.health),
        })
    columns = [
        ("label", "组 / 团队 / Lead"),
        ("members", "人数"),
        ("covered", "覆盖人数"),
        ("utilisation", "使用率(%)"),
        ("vs_company", "对比公司(%)"),
        ("low_utilisers", "低使用率人数"),
        ("entitled", "应得(天)"),
        ("used", "已用(天)"),
        ("forfeit", "临期未休(天)"),
        ("comp", "调休堆积(天)"),
        ("health", "健康度"),
    ]
    return await xlsx_export_response(
        db, request, current_user,
        columns=columns, rows=rows, total_matched=len(rows),
        filename_stem=f"leave_insights_{dimension}",
        sheet_title="休假健康洞察",
        audit_action="hr.leave_insights.export",
        audit_resource_type="hr_leave_insights",
        filters={"dimension": dimension},
    )