"""Per-person lifecycle flow view — the admin API behind
``/admin/lifecycle/person/{id}``.

Aggregates everything across the hr / offboarding / provisioning /
notification / compliance domains into a single payload so the flow-
diagram UI can render without N+1 fetches. Read-only.
"""
from __future__ import annotations

from datetime import UTC, datetime
from uuid import UUID

from fastapi import APIRouter, Depends, Request
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.auth.models import AuthUser
from app.compliance.models import (
    INCIDENT_OPEN,
    ComplianceIncident,
)
from app.core.database import get_db
from app.core.dependencies import get_current_user, require_admin_read
from app.core.exceptions import ForbiddenError, NotFoundError
from app.core.response import success_response
from app.domains.hr.models import HrLifecycleEvent
from app.domains.offboarding.models import (
    OffboardingBlocker,
    OffboardingProcess,
    OffboardingReminderLog,
)
from app.domains.provisioning.models import ProvisioningTask
from app.identity.person_models import IdentityPerson
from app.notifications.models import NotificationOutbox

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


def _iso(dt: datetime | None) -> str | None:
    return dt.isoformat() if dt else None


@router.get(
    "/me",
    summary="Current user's lifecycle flow — employee-facing self view",
)
async def my_flow(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Employee-facing `/my/onboarding` | `/my/offboarding` payload.

    Same shape as ``/person/{id}`` but scoped to the calling user's
    own id; no admin role required. The UI hides admin-only sections
    (compliance incidents, raw notification history) on this route.
    """
    return await _person_flow_payload(
        person_id=current_user.id, request=request, db=db, is_admin=False,
    )


def _require_forensic_read(user: AuthUser) -> None:
    """A per-person case file is not leadership-tier reading.

    ``require_admin_read`` admits ``management`` because most of what it
    guards is governance metadata — process lists, job catalogues, task
    queues. These two routes are different in kind: they aggregate one
    NAMED person's history across eight domains into a single case file,
    which is precisely what the auditor domain narrowed its frozen
    evidence bundles to ``global_admin`` / ``auditor`` for. Leaving the
    identical data reachable here would have made that narrowing a door
    that only looks locked.
    """
    from app.core.rbac import can_manage_security  # noqa: PLC0415

    if not can_manage_security(user.role_names):
        raise ForbiddenError(
            "auditor or global_admin role required to read a person's "
            "forensic case file"
        )


@router.get(
    "/person/{person_id}",
    summary="Per-person lifecycle flow payload (admin flow diagram)",
)
async def person_flow(
    person_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    return await _person_flow_payload(
        person_id=person_id, request=request, db=db,
    )


@router.get(
    "/forensic/{person_id}",
    summary="Forensic timeline — per-person case file across 8 domains",
)
async def forensic_timeline(
    person_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Aggregated time-desc timeline + 4 KPIs.

    Reads — but does not filter on — ``IdentityPerson.deleted_at``, so
    soft-deleted leavers are queryable for retrospective evidence work.
    The companion ZIP export is at
    ``/lifecycle/forensic/{person_id}/evidence-pack``.
    """
    _require_forensic_read(current_user)
    from app.hr_lifecycle.forensic_service import ForensicTimelineService

    svc = ForensicTimelineService(db)
    person = await svc.get_person(person_id)
    if person is None:
        raise NotFoundError("IdentityPerson", str(person_id))
    payload = await svc.build_payload(person)
    return success_response(data=payload, request=request)


@router.get(
    "/forensic/{person_id}/evidence-pack",
    summary="Forensic evidence pack — ZIP download (admin / auditor)",
)
async def forensic_evidence_pack(
    person_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """Return the same forensic payload as ``/forensic/{person_id}``
    packaged as a ZIP (manifest.json + timeline.json + timeline.csv +
    README.txt). Every download is audit-logged with the caller's
    user id so the export trail itself is recoverable later.
    """
    _require_forensic_read(current_user)
    from uuid import uuid4

    from fastapi import Response

    from app.audit.service import AuditService
    from app.hr_lifecycle.forensic_service import (
        ForensicTimelineService,
        build_evidence_pack_bytes,
    )

    svc = ForensicTimelineService(db)
    person = await svc.get_person(person_id)
    if person is None:
        raise NotFoundError("IdentityPerson", str(person_id))
    # Evidence-pack path uses a much higher per-source cap (5000) than
    # the interactive JSON endpoint (500). The legal/auditor export
    # needs to genuinely cover a 10-year-tenure case file — the JSON
    # cap stays low because the browser timeline doesn't paginate.
    # 5000 still bounds resource_id audit scans to a tolerable range.
    payload = await svc.build_payload(person, per_source_limit=5000)
    zip_bytes = build_evidence_pack_bytes(payload, person)

    # Use the request's x-trace-id if present, else mint a fresh UUID
    # per download — never coalesce to a constant string, otherwise
    # every header-less download lands on the same synthetic trace_id
    # and the audit story loses per-export distinguishability.
    trace_header = request.headers.get("x-trace-id")
    trace_id_value = trace_header or str(uuid4())

    audit = AuditService(db)
    await audit.log(
        trace_id=trace_id_value,
        actor_type="user",
        actor_user_id=current_user.id,
        action="lifecycle.forensic.evidence_pack_downloaded",
        resource_type="identity_person",
        resource_id=str(person.id),
        status="ok",
        detail={
            "row_count": payload["row_count"],
            "per_source_limit": payload["per_source_limit"],
            "truncated_by_source": payload.get("truncated_by_source", {}),
            "kpis": payload["kpis"],
        },
        ip_address=request.client.host if request.client else None,
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    filename = f"forensic-{person.id}.zip"
    return Response(
        content=zip_bytes,
        media_type="application/zip",
        headers={
            "Content-Disposition": f'attachment; filename="{filename}"',
            "Content-Length": str(len(zip_bytes)),
        },
    )


async def _person_flow_payload(
    *,
    person_id: UUID,
    request: Request,
    db: AsyncSession,
    is_admin: bool = True,
):
    # Eager-load ``manager`` so the payload builder can read
    # ``person.manager.display_name`` without triggering a lazy-load
    # (MissingGreenlet in async context — the 500 that was hitting
    # /admin/lifecycle/person/{id} for any employee whose
    # ``manager_person_id`` was non-NULL).
    person = (
        await db.execute(
            select(IdentityPerson)
            .options(selectinload(IdentityPerson.manager))
            .where(IdentityPerson.id == person_id)
        )
    ).scalar_one_or_none()
    if person is None or person.deleted_at is not None:
        raise NotFoundError("IdentityPerson", str(person_id))

    # HRIS lifecycle events (hire/offboard)
    events = list((
        await db.execute(
            select(HrLifecycleEvent)
            .where(HrLifecycleEvent.person_id == person_id)
            .order_by(HrLifecycleEvent.effective_date.asc(),
                      HrLifecycleEvent.created_at.asc())
        )
    ).scalars().all())

    # Provisioning tasks
    tasks = list((
        await db.execute(
            select(ProvisioningTask)
            .where(ProvisioningTask.user_id == person_id)
            .order_by(ProvisioningTask.created_at.asc())
        )
    ).scalars().all())

    # Pre-launch P2 fix (Round 1, CR-011): resolve software_name in a
    # batch SELECT. The prior ``getattr(t, "software_name", None)``
    # always returned None because ``ProvisioningTask`` has no such
    # column — only ``software_id``. Frontend rendered a blank column
    # for every row.
    from app.domains.asset.models import AssetSoftwareCatalog  # noqa: PLC0415
    _software_ids = {t.software_id for t in tasks if t.software_id}
    _software_name_by_id: dict[UUID, str] = {}
    if _software_ids:
        _sw_rows = (
            await db.execute(
                select(
                    AssetSoftwareCatalog.id, AssetSoftwareCatalog.name,
                ).where(AssetSoftwareCatalog.id.in_(_software_ids))
            )
        ).all()
        _software_name_by_id = {row[0]: row[1] for row in _sw_rows}

    # Offboarding process + blockers + reminders (if any)
    process = await db.scalar(
        select(OffboardingProcess).where(
            OffboardingProcess.person_id == person_id
        )
    )
    blockers: list = []
    reminders: list = []
    if process is not None:
        blockers = list((
            await db.execute(
                select(OffboardingBlocker)
                .where(OffboardingBlocker.process_id == process.id)
                .order_by(OffboardingBlocker.created_at.asc())
            )
        ).scalars().all())
        reminders = list((
            await db.execute(
                select(OffboardingReminderLog)
                .where(OffboardingReminderLog.process_id == process.id)
                .order_by(OffboardingReminderLog.sent_at.asc())
            )
        ).scalars().all())

    # Compliance incidents tied to this person
    incidents = list((
        await db.execute(
            select(ComplianceIncident)
            .where(ComplianceIncident.person_id == person_id)
            .order_by(ComplianceIncident.detected_at.desc())
        )
    ).scalars().all())

    # Notification history (last 20) from outbox
    notifications = list((
        await db.execute(
            select(NotificationOutbox)
            .where(NotificationOutbox.recipient_person_id == person_id)
            .order_by(NotificationOutbox.created_at.desc())
            .limit(20)
        )
    ).scalars().all())

    # Flow computed status per phase so the UI can paint a timeline
    # without embedding business logic.
    flow = _compute_flow(
        person=person,
        events=events,
        tasks=tasks,
        process=process,
        blockers=blockers,
        reminders=reminders,
        incidents=incidents,
    )

    payload = {
        "person": {
            "id": str(person.id),
            "display_name": person.display_name,
            "work_email": person.work_email,
            "employee_no": person.employee_no,
            "employment_status": person.employment_status,
            "hire_date": person.hire_date.isoformat() if person.hire_date else None,
            "termination_date": (
                person.termination_date.isoformat()
                if person.termination_date else None
            ),
            "department": person.primary_department,
            "job_title": person.job_title,
            "manager_person_id": (
                str(person.manager_person_id) if person.manager_person_id else None
            ),
            "manager_name": (
                person.manager.display_name if person.manager is not None else None
            ),
            "manager_employee_no": (
                person.manager.employee_no if person.manager is not None else None
            ),
            "manager_employment_status": (
                person.manager.employment_status if person.manager is not None else None
            ),
            "it_info_sent_at": _iso(person.it_info_sent_at),
            "last_synced_at": _iso(person.last_synced_at),
        },
        "flow": flow,
        "hris_events": [
            {
                "id": str(e.id),
                "event_type": e.event_type,
                "effective_date": e.effective_date.isoformat(),
                "detail": e.detail,
                "created_at": _iso(e.created_at),
            }
            for e in events
        ],
        "provisioning_tasks": [
            {
                "id": str(t.id),
                "software_id": str(t.software_id) if t.software_id else None,
                "software_name": (
                    _software_name_by_id.get(t.software_id)
                    if t.software_id else None
                ),
                "action": t.action,
                "execution_mode": t.execution_mode,
                "status": t.status,
                "source": t.source,
                "assignee_id": str(t.assignee_id) if t.assignee_id else None,
                "created_at": _iso(t.created_at),
                "completed_at": _iso(t.completed_at),
                "error_message": t.error_message,
            }
            for t in tasks
        ],
        "offboarding": (
            {
                "process": {
                    "id": str(process.id),
                    "status": process.status,
                    "leaver_date": process.leaver_date.isoformat(),
                    "scanned_at": _iso(process.scanned_at),
                    "ready_at": _iso(process.ready_at),
                    "sso_suspended_at": _iso(process.sso_suspended_at),
                    "completed_at": _iso(process.completed_at),
                },
                "blockers": [
                    {
                        "id": str(b.id),
                        "resource_type": b.resource_type,
                        "resource_id": (
                            str(b.resource_id) if b.resource_id else None
                        ),
                        "label": b.label,
                        "resolution": b.resolution,
                        "resolution_ref": b.resolution_ref,
                        "notes": b.notes,
                    }
                    for b in blockers
                ],
                "reminders": [
                    {
                        "id": str(r.id),
                        "reminder_kind": r.reminder_kind,
                        "sent_at": _iso(r.sent_at),
                        "channels": r.channels,
                        # Round-13: expose ``detail`` so the lifecycle
                        # admin Modal can show recipients, blocker_ids,
                        # and per-tick context (manager/IT id, why a
                        # T-7/T-3/T-1 was a no-op, etc.) without an
                        # extra API call.
                        "detail": r.detail,
                    }
                    for r in reminders
                ],
            }
            if process is not None else None
        ),
        "compliance_incidents": [
            {
                "id": str(i.id),
                "incident_type": i.incident_type,
                "severity": i.severity,
                "status": i.status,
                "title": i.title,
                "resource_type": i.resource_type,
                "resource_label": i.resource_label,
                "detected_at": _iso(i.detected_at),
                "resolved_at": _iso(i.resolved_at),
            }
            for i in incidents
        ],
        "recent_notifications": [
            {
                "id": str(n.id),
                "title": n.title,
                # Round-13: expose body / url / per-channel outcome /
                # last_error so the lifecycle admin Modal can render the
                # full notification without round-tripping for detail.
                # ``body_text`` is templated text; long, but truncating
                # for the admin would defeat the diagnostic point.
                "body_text": n.body_text,
                "url": n.url,
                "template_key": n.template_key,
                "status": n.status,
                "channels": n.channels,
                "delivered_channels": n.delivered_channels,
                "failed_channels": n.failed_channels,
                "last_error": n.last_error,
                "attempts": n.attempts,
                "max_attempts": n.max_attempts,
                "created_at": _iso(n.created_at),
                "delivered_at": _iso(n.delivered_at),
                "next_attempt_at": _iso(n.next_attempt_at),
            }
            for n in notifications
        ],
    }
    if not is_admin:
        payload["compliance_incidents"] = []
        payload["recent_notifications"] = []
    return success_response(data=payload, request=request)


def _compute_flow(
    *,
    person: IdentityPerson,
    events: list[HrLifecycleEvent],
    tasks: list[ProvisioningTask],
    process: OffboardingProcess | None,
    blockers: list[OffboardingBlocker],
    reminders: list[OffboardingReminderLog],
    incidents: list[ComplianceIncident],
) -> dict:
    """Compute a list of flow nodes the UI renders left-to-right.

    Each node has ``{key, label, status, detail, timestamp}`` where
    ``status ∈ {done, active, blocked, pending, skipped}``.
    Onboarding persons get the onboarding track; leavers get the
    offboarding track. An ``active`` person who already has an
    offboarding process is rendered in the offboarding track — the
    platform's dominant concern is always "is this person's exit
    clean?". This must key on the process itself, not just
    ``termination_date``: an HRIS offboard webhook lands days before
    the SF status flips (P0119, 2026-07-31), so the process exists
    while the person is still ``active`` with no termination_date —
    ``_upsert_person`` deliberately keeps them active while SF says
    ``status='t'``.
    """
    today = datetime.now(UTC).date()

    # #2234 — ``OffboardingProcess.person_id`` is UNIQUE, so a person has
    # exactly ONE process row for life. A previous cycle's process stuck
    # in a non-terminal state (``scanning`` / ``blocked``) is a proven
    # production shape (15 permanently-pending orphan transfer orders
    # pinned processes at ``blocked``), and nothing in the onboard /
    # reactivate path terminates it. Without this carve-out such a person,
    # once REHIRED, renders the offboarding step chain for ever and their
    # actual onboarding readiness chain is hidden. The discriminator is
    # the employment cycle: a process created BEFORE the current
    # ``hire_date`` belongs to the previous one. The P0119 shape this
    # branch exists for is untouched — there the process is created
    # days AFTER the (original, years-old) hire_date.
    proc_created = getattr(process, "created_at", None) if process else None
    if proc_created is not None and hasattr(proc_created, "date"):
        proc_created = proc_created.date()
    stale_prior_cycle = (
        process is not None
        and person.employment_status not in ("leaver_pending", "left")
        and person.termination_date is None
        and person.hire_date is not None
        and proc_created is not None
        and proc_created < person.hire_date
    )

    # Decide track
    if not stale_prior_cycle and (
        person.employment_status in ("leaver_pending", "left")
        or person.termination_date is not None
        or (
            process is not None
            and process.status not in ("completed", "cancelled")
        )
    ):
        return _compute_offboarding_flow(
            person=person,
            process=process,
            blockers=blockers,
            reminders=reminders,
            incidents=incidents,
            today=today,
        )
    return _compute_onboarding_flow(
        person=person,
        events=events,
        tasks=tasks,
        today=today,
    )


def _compute_onboarding_flow(
    *,
    person: IdentityPerson,
    events: list[HrLifecycleEvent],
    tasks: list[ProvisioningTask],
    today,
) -> dict:
    nodes: list[dict] = []

    # Node 1 — HRIS sync
    onboard_events = [e for e in events if e.event_type == "onboard"]
    synced = bool(onboard_events)
    nodes.append({
        "key": "hris_sync",
        "label": "HRIS 同步",
        "status": "done" if synced else "pending",
        "timestamp": (
            _iso(onboard_events[0].created_at) if synced else None
        ),
        "detail": (
            f"{len(onboard_events)} 条 onboard 事件"
            if synced else "尚未同步"
        ),
    })

    # Node 2 — Provisioning
    task_done = sum(1 for t in tasks if t.status in ("completed",))
    task_total = len(tasks)
    task_failed = sum(1 for t in tasks if t.status == "failed")
    if task_total == 0:
        prov_status = "pending"
        prov_detail = "暂无策略命中"
    elif task_failed > 0:
        prov_status = "blocked"
        prov_detail = f"{task_done}/{task_total} 完成 · {task_failed} 失败"
    elif task_done == task_total:
        prov_status = "done"
        prov_detail = f"{task_done}/{task_total} 完成"
    else:
        prov_status = "active"
        prov_detail = f"{task_done}/{task_total} 完成"
    nodes.append({
        "key": "provisioning",
        "label": "账号/权限发放",
        "status": prov_status,
        "detail": prov_detail,
    })

    # Node 3 — Hire-day IT info push
    if person.it_info_sent_at is not None:
        nodes.append({
            "key": "it_info_push",
            "label": "IT 信息推送",
            "status": "done",
            "timestamp": _iso(person.it_info_sent_at),
            "detail": f"通道 {', '.join(person.it_info_sent_channels or [])}",
        })
    elif person.hire_date is None:
        nodes.append({
            "key": "it_info_push",
            "label": "IT 信息推送",
            "status": "pending",
            "detail": "未设置入职日",
        })
    elif person.hire_date > today:
        nodes.append({
            "key": "it_info_push",
            "label": "IT 信息推送",
            "status": "pending",
            "detail": f"计划于 {person.hire_date.isoformat()} 09:00 推送",
        })
    else:
        nodes.append({
            "key": "it_info_push",
            "label": "IT 信息推送",
            "status": "blocked",
            "detail": "入职日已到但未推送",
        })

    # Node 4 — Active
    nodes.append({
        "key": "active",
        "label": "激活就绪",
        "status": (
            "done" if person.employment_status == "active" else "pending"
        ),
        "detail": f"当前状态: {person.employment_status}",
    })

    return {
        "track": "onboarding",
        "nodes": nodes,
        "next_action": _onboarding_next_action(person, tasks, today),
    }


def _onboarding_next_action(
    person: IdentityPerson, tasks: list[ProvisioningTask], today,
) -> dict | None:
    if person.it_info_sent_at is None and person.hire_date and person.hire_date >= today:
        return {
            "key": "hire_day_notify",
            "description": "入职当天 09:00 自动推送 IT 信息",
            "scheduled_at": f"{person.hire_date.isoformat()}T09:00+08:00",
        }
    pending_tasks = [t for t in tasks if t.status in ("pending", "in_progress")]
    if pending_tasks:
        return {
            "key": "provisioning_dispatch",
            "description": f"{len(pending_tasks)} 项发放任务待执行",
        }
    return None


def _compute_offboarding_flow(
    *,
    person: IdentityPerson,
    process: OffboardingProcess | None,
    blockers: list[OffboardingBlocker],
    reminders: list[OffboardingReminderLog],
    incidents: list[ComplianceIncident],
    today,
) -> dict:
    nodes: list[dict] = []

    # Node 1 — HRIS (offboarding event recorded)
    nodes.append({
        "key": "hris_sync",
        "label": "HRIS 同步",
        "status": "done" if process is not None else "pending",
        "timestamp": _iso(process.scanned_at) if process else None,
        "detail": (
            f"预计离职: {person.termination_date.isoformat()}"
            if person.termination_date else "未设置 termination_date"
        ),
    })

    # Node 2 — blockers scan
    if process is None:
        nodes.append({
            "key": "blocker_scan",
            "label": "资源扫描",
            "status": "pending",
            "detail": "尚未启动流程",
        })
    else:
        pending = sum(1 for b in blockers if b.resolution == "pending")
        total = len(blockers)
        if total == 0:
            nodes.append({
                "key": "blocker_scan",
                "label": "资源扫描",
                "status": "done",
                "timestamp": _iso(process.scanned_at),
                "detail": "无挂钩资源",
            })
        elif pending == 0:
            nodes.append({
                "key": "blocker_scan",
                "label": "资源扫描",
                "status": "done",
                "timestamp": _iso(process.scanned_at),
                "detail": f"{total} 项已全部处理",
            })
        else:
            nodes.append({
                "key": "blocker_scan",
                "label": "资源扫描",
                "status": "active",
                "timestamp": _iso(process.scanned_at),
                "detail": f"{pending}/{total} 项待处理",
            })

    # Node 3 — reminders
    reminder_count = len(reminders)
    nodes.append({
        "key": "reminders",
        "label": "提醒推送",
        "status": "done" if reminder_count > 0 else "pending",
        "detail": f"已发送 {reminder_count} 条" if reminder_count > 0 else "尚未触发",
    })

    # Node 4 — T+0 force transfer
    if process and process.ready_at is not None:
        nodes.append({
            "key": "force_transfer",
            "label": "T+0 强制转让",
            "status": "done",
            "timestamp": _iso(process.ready_at),
            "detail": "所有资源已迁移",
        })
    elif person.termination_date and person.termination_date <= today:
        nodes.append({
            "key": "force_transfer",
            "label": "T+0 强制转让",
            "status": "active",
            "detail": "每 15 分钟自动尝试,直至清零",
        })
    else:
        nodes.append({
            "key": "force_transfer",
            "label": "T+0 强制转让",
            "status": "pending",
            "detail": (
                f"预定 {person.termination_date.isoformat()}"
                if person.termination_date else "未设置离职日"
            ),
        })

    # Node 5 — SSO suspend
    if process and process.sso_suspended_at:
        nodes.append({
            "key": "sso_suspend",
            "label": "SSO 停用",
            "status": "done",
            "timestamp": _iso(process.sso_suspended_at),
        })
    else:
        nodes.append({
            "key": "sso_suspend",
            "label": "SSO 停用",
            "status": "pending",
        })

    # Node 6 — Left
    nodes.append({
        "key": "left",
        "label": "状态转为 Left",
        "status": "done" if person.employment_status == "left" else "pending",
        "detail": f"当前状态: {person.employment_status}",
    })

    # Node 7 — T+1 verification
    open_incidents = [i for i in incidents if i.status == INCIDENT_OPEN]
    if person.employment_status == "left":
        if open_incidents:
            nodes.append({
                "key": "post_verify",
                "label": "T+1 复查",
                "status": "blocked",
                "detail": f"{len(open_incidents)} 项合规告警,请处理",
            })
        else:
            nodes.append({
                "key": "post_verify",
                "label": "T+1 复查",
                "status": "done",
                "detail": "无残留资源",
            })
    else:
        nodes.append({
            "key": "post_verify",
            "label": "T+1 复查",
            "status": "pending",
            "detail": "Left 状态次日自动复查",
        })

    return {
        "track": "offboarding",
        "nodes": nodes,
        "next_action": _offboarding_next_action(
            person=person, process=process, blockers=blockers,
            incidents=incidents, today=today,
        ),
    }


def _offboarding_next_action(
    *,
    person: IdentityPerson,
    process: OffboardingProcess | None,
    blockers: list[OffboardingBlocker],
    incidents: list[ComplianceIncident],
    today,
) -> dict | None:
    open_incidents = [i for i in incidents if i.status == INCIDENT_OPEN]
    if open_incidents:
        return {
            "key": "resolve_incidents",
            "description": f"{len(open_incidents)} 项合规告警待人工处理",
        }
    if person.employment_status == "left":
        return None
    pending = [b for b in blockers if b.resolution == "pending"]
    if not pending:
        return {
            "key": "waiting_tick",
            "description": "待定时任务 hr.process_advance 推进状态(每 15 分钟)",
        }
    if person.termination_date and person.termination_date <= today:
        return {
            "key": "force_transfer_tick",
            "description": f"下一个 15 分钟 tick 将强制转让 {len(pending)} 项资源",
        }
    return {
        "key": "awaiting_leaver_date",
        "description": (
            f"{len(pending)} 项待处理,leaver_date ({person.termination_date}) "
            "到来后将触发强制转让"
        ),
    }