"""Provisioning admin endpoints.

All endpoints require IT Admin role; some also accept Asset Admin.
"""
import html
import logging
from uuid import UUID

import jwt
from fastapi import APIRouter, Depends, Form, Query, Request
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    enforce_action_token_subject,
    get_current_user,
    require_active_employee_for_mutations,
    require_admin,
    require_hr_admin,
    require_lifecycle_read,
)
from app.core.exceptions import AppException, ForbiddenError
from app.core.idempotent_route import idempotent
from app.core.person_scope import SubordinateScope, subordinate_person_ids
from app.core.rbac import (
    GLOBAL_ADMIN_ROLE,
    HR_ROLE,
    MANAGEMENT_ROLE,
    has_role,
)
from app.core.response import paginated_response, success_response
from app.core.security import decode_action_token
from app.core.working_hours import (
    PROVISION_LEAD_HOURS_MAX,
    PROVISION_LEAD_HOURS_MIN,
)
from app.domains.provisioning.bundle_service import BundleService
from app.domains.provisioning.models import (
    ACTION_TOKEN_PURPOSE_MANUAL_COMPLETE,
    EXEC_MANUAL_NOTIFY,
    SOURCE_ASSET_DISPOSAL,
    TARGET_SSO_GRANT,
)
from app.domains.provisioning.schemas import (
    BulkExecuteRequest,
    ProvisionPolicyCreate,
    ProvisionPolicyUpdate,
    ReconcileApplyRequest,
    SoftwareConnectorCreate,
    SoftwareConnectorUpdate,
    TaskCompleteRequest,
    TaskFailRequest,
    TaskRetryRequest,
)
from app.domains.provisioning.service import ProvisioningService

# Leaver-runtime gate: matches permissions/asset routers. A user whose
# offboarding has flipped employment_status into the leaver-pending
# window cannot mutate provisioning state — including assignee
# completion of in-flight tasks. Read endpoints (`get_current_user`)
# stay accessible so the IT 宇宙 page still loads for a person mid-
# offboarding.
logger = logging.getLogger(__name__)

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


# ─── Connector driver catalog (software_provisioning-kind drivers) ───


@router.get(
    "/connector-types",
    summary="List available software-provisioning drivers (for the binding form)",
)
async def list_connector_types(
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Returns the same metadata shape as ``/api/v1/integrations/kinds``
    filtered to ``kind='software_provisioning'`` — exposes each driver's
    tenant-level ``fields`` and per-software ``binding_fields`` so the
    admin UI can render the right forms."""
    svc = ProvisioningService(db)
    return success_response(
        data=[m.model_dump() for m in svc.list_available_connector_types()],
        request=request,
    )


# ─── Connector CRUD ───


@router.get("/connectors")
async def list_connectors(
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    connectors = await svc.list_connectors()
    return success_response(data=connectors, request=request)


@router.get("/onboarding-overview")
async def onboarding_overview(
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Account-creation config snapshot for the onboarding-strategy page —
    Google ``auto_create_on_hire`` + Feishu connector ``operation`` mode, so
    the page can show the full onboarding picture (build-account switches +
    seat/permission policies) in one place. Read-only."""
    svc = ProvisioningService(db)
    overview = await svc.onboarding_account_overview()
    return success_response(data=overview, request=request)


# ─── Onboarding provisioning lead window (HR-configurable) ───


class OnboardingLeadSettingsBody(BaseModel):
    """How many hours BEFORE a hire's local start date their accounts
    (Google + Feishu) are created and the Day-1 welcome/password mails go
    out. Bounds mirror ``app.core.working_hours``: 0 = strictly on the
    local hire date; 168 (one week) is the ceiling — anything longer
    re-opens the "live mailbox + plaintext-password mail sitting
    unattended for weeks" failure mode."""
    provision_lead_hours: int = Field(
        ge=PROVISION_LEAD_HOURS_MIN, le=PROVISION_LEAD_HOURS_MAX,
    )


@router.get(
    "/onboarding-lead-settings",
    summary="Read the onboarding provisioning lead window (hours)",
)
async def get_onboarding_lead_settings(
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    from app.core.working_hours import (
        DEFAULT_PROVISION_LEAD_HOURS,
        PROVISION_LEAD_HOURS_MAX,
        PROVISION_LEAD_HOURS_MIN,
        SETTING_PROVISION_LEAD_HOURS,
        _coerce_lead_hours,
    )
    from app.integrations.service import get_platform_setting

    raw = await get_platform_setting(
        SETTING_PROVISION_LEAD_HOURS,
        default=DEFAULT_PROVISION_LEAD_HOURS,
        session=db,
    )
    hours = _coerce_lead_hours(raw)
    return success_response(
        data={
            "provision_lead_hours": (
                hours if hours is not None else DEFAULT_PROVISION_LEAD_HOURS
            ),
            "default_hours": DEFAULT_PROVISION_LEAD_HOURS,
            "min_hours": PROVISION_LEAD_HOURS_MIN,
            "max_hours": PROVISION_LEAD_HOURS_MAX,
        },
        request=request,
    )


@router.put(
    "/onboarding-lead-settings",
    summary="Update the onboarding provisioning lead window (hours)",
)
async def update_onboarding_lead_settings(
    body: OnboardingLeadSettingsBody,
    request: Request,
    current_user: AuthUser = Depends(require_hr_admin()),
    db: AsyncSession = Depends(get_db),
):
    """HR + global_admin editable (``require_hr_admin``): the lead window
    is an HR-owned onboarding policy, not an integration credential. Every
    consumer (HRIS confirm hooks, identity-sync sweeps, the Google account
    minter, the hourly hire-day cron) resolves the value per flow, so a
    change takes effect on the next tick without a deploy."""
    from app.core.working_hours import (
        SETTING_PROVISION_LEAD_HOURS,
        set_provision_lead_hours_cache,
    )
    from app.integrations.service import PlatformSettingService

    svc = PlatformSettingService(db)
    await svc.upsert(
        SETTING_PROVISION_LEAD_HOURS,
        body.provision_lead_hours,
        "Onboarding account-provisioning lead window in hours (Google + "
        "Feishu creation and Day-1 welcome mails open this many hours "
        "before the hire's local start date)",
        trace_id=str(request.state.trace_id),
        actor_user_id=current_user.id,
    )
    await db.commit()
    # The writing process's own gates see the new value immediately;
    # sibling processes (worker) converge within the 60s TTL.
    set_provision_lead_hours_cache(body.provision_lead_hours)
    return success_response(data=body.model_dump(), request=request)


@router.post("/connectors")
async def create_connector(
    request: Request,
    body: SoftwareConnectorCreate,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    connector = await svc.create_connector(
        body, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=connector, request=request)


@router.patch("/connectors/{connector_id}")
async def update_connector(
    request: Request,
    connector_id: UUID,
    body: SoftwareConnectorUpdate,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    connector = await svc.update_connector(
        connector_id, body, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=connector, request=request)


@router.post("/connectors/{connector_id}/test")
async def test_connector(
    request: Request,
    connector_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Invoke the connector's test_connection to validate credentials."""
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    success, error, raw = await svc.test_connector(
        connector_id, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(
        data={"success": success, "error_message": error, "raw_response": raw},
        request=request,
    )


@router.delete("/connectors/{connector_id}")
async def delete_connector(
    request: Request,
    connector_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    await svc.delete_connector(connector_id, trace_id=trace_id, actor_user_id=current_user.id)
    await db.commit()
    return success_response(data=None, request=request)


# ─── Provisioning tasks (admin queue) ───


@router.get("/tasks")
async def list_tasks(
    request: Request,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    status: str | None = Query(None),
    statuses: str | None = Query(
        None,
        description="Comma-separated statuses; takes precedence over `status`",
    ),
    source: str | None = Query(None),
    user_id: UUID | None = Query(None),
    software_id: UUID | None = Query(None),
    assignee_id: UUID | None = Query(None),
    action: str | None = Query(
        None, description="grant | revoke",
    ),
    execution_modes: str | None = Query(
        None,
        description="Comma-separated execution_mode values "
        "(auto_api,manual_notify,sso_auto)",
    ),
    keyword: str | None = Query(
        None,
        max_length=100,
        description=(
            "Case-insensitive substring match against the target "
            "person's display_name / work_email and the software's "
            "catalog name. Powers the Software Cancellations search "
            "box across the auto-done / manual-pending / failed "
            "segments. Empty / whitespace-only values are ignored. "
            "LIKE metacharacters (%, _, \\) are escaped server-side "
            "so they match literally. Capped at 100 chars to bound "
            "the worst-case LIKE-pattern scan."
        ),
    ),
    order_by: str | None = Query(
        None,
        description=(
            "Sort key. Default is ``created_at`` (newest enqueue "
            "first). Pass ``completed_at`` for the cancellations "
            "overview's Completed segment so rows that finished most "
            "recently surface first; NULL completed_at values land "
            "at the bottom. Unknown values silently fall back to "
            "``created_at`` — no 400."
        ),
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List provisioning tasks.

    Global admin + management see everything (management is the
    read-only leadership view). Employees may only request their own
    slice (``user_id=<self>``) — the IT 宇宙 page on /me uses this to
    surface in-flight grant / revoke jobs the platform is running on
    their behalf. Cross-user filters from a non-admin caller are 403.

    The ``statuses`` (multi-value) and ``execution_modes`` (multi-value)
    parameters power the Software Cancellations admin overview, where a
    single segment needs to combine e.g. ``status IN (pending,notified)``
    or ``execution_mode IN (auto_api,sso_auto)`` without N round-trips.

    2026-07-15: ``hr`` joins global_admin + management on the view-all
    path so the 任务队列 page (added to the HR 入离职管理 menu this PR)
    renders the full queue read-only. Mutations (execute / cancel /
    fail / bulk) stay ``require_admin`` — hr inspects, never acts.
    """
    if not has_role(
        current_user.role_names, GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE, HR_ROLE,
    ):
        if user_id is None or user_id != current_user.id:
            raise ForbiddenError(
                "You may only view your own provisioning tasks.",
            )
    # Strip whitespace so an MCP / curl caller typing the natural
    # "a, b" form doesn't silently match zero rows because " b" never
    # equals "b" in the SQL IN (...) clause.
    statuses_list = (
        [s.strip() for s in (statuses or "").split(",") if s.strip()] or None
    )
    exec_modes_list = (
        [m.strip() for m in (execution_modes or "").split(",") if m.strip()] or None
    )
    svc = ProvisioningService(db)
    tasks, total = await svc.list_tasks(
        page=page,
        page_size=page_size,
        status=status,
        statuses=statuses_list,
        source=source,
        user_id=user_id,
        software_id=software_id,
        assignee_id=assignee_id,
        action=action,
        execution_modes=exec_modes_list,
        keyword=keyword,
        order_by=order_by,
    )
    return paginated_response(
        data=tasks, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/tasks/export.xlsx")
async def export_tasks(
    request: Request,
    status: str | None = Query(None),
    statuses: str | None = Query(None),
    source: str | None = Query(None),
    user_id: UUID | None = Query(None),
    software_id: UUID | None = Query(None),
    assignee_id: UUID | None = Query(None),
    action: str | None = Query(None),
    execution_modes: str | None = Query(None),
    keyword: str | None = Query(None, max_length=100),
    order_by: str | None = Query(None),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Export the filtered provisioning queue as XLSX.

    Filter signature mirrors :func:`list_tasks` minus pagination — the
    UI passes the exact query params the queue page is showing, so the
    download matches the screen. RBAC mirrors the list endpoint too:
    non-admin callers may only export their own slice (hr / management
    join global_admin on the view-all path — 2026-07-15). Reuses
    ``ProvisioningService.list_tasks`` so any filter fix on the list
    applies here automatically. Capped at
    :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows.
    """
    from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response

    if not has_role(
        current_user.role_names, GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE, HR_ROLE,
    ):
        if user_id is None or user_id != current_user.id:
            raise ForbiddenError(
                "You may only export your own provisioning tasks.",
            )
    statuses_list = (
        [s.strip() for s in (statuses or "").split(",") if s.strip()] or None
    )
    exec_modes_list = (
        [m.strip() for m in (execution_modes or "").split(",") if m.strip()] or None
    )
    svc = ProvisioningService(db)
    tasks, total = await svc.list_tasks(
        page=1,
        page_size=EXPORT_ROW_LIMIT,
        status=status,
        statuses=statuses_list,
        source=source,
        user_id=user_id,
        software_id=software_id,
        assignee_id=assignee_id,
        action=action,
        execution_modes=exec_modes_list,
        keyword=keyword,
        order_by=order_by,
    )
    columns = [
        ("user_name", "人员"),
        ("user_email", "邮箱"),
        ("software_name", "软件"),
        ("target_label", "对象"),
        ("target_type", "对象类型"),
        ("action", "动作"),
        ("source", "来源"),
        ("execution_mode", "执行方式"),
        ("status", "状态"),
        ("assignee_name", "处理人"),
        ("attempt_count", "尝试次数"),
        ("error_message", "错误信息"),
        ("external_ref", "外部引用"),
        ("not_before", "不早于"),
        ("started_at", "开始时间"),
        ("completed_at", "完成时间"),
        ("created_at", "创建时间"),
        ("id", "任务ID"),
    ]
    rows = [t.model_dump() for t in tasks]
    return await xlsx_export_response(
        db,
        request,
        current_user,
        columns=columns,
        rows=rows,
        total_matched=total,
        filename_stem="provisioning_tasks",
        sheet_title="Provisioning Tasks",
        audit_action="provisioning.tasks.export",
        audit_resource_type="provisioning_task",
        filters={
            "status": status,
            "statuses": statuses,
            "source": source,
            "user_id": user_id,
            "software_id": software_id,
            "assignee_id": assignee_id,
            "action": action,
            "execution_modes": execution_modes,
            "keyword": keyword,
        },
    )


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

    Admin-read gated (global_admin + management). Returns KPI totals +
    daily created-vs-completed trend + status/action/execution_mode/
    source/target_type rollups — replacing the page's former 4× narrow
    count probes. Read-only augment; no MCP tool (mirrors the existing
    定时任务/审计/审批 analytics tabs).
    """
    svc = ProvisioningService(db)
    data = await svc.analytics_summary(window_days=window_days)
    return success_response(data=data, request=request)


@router.get("/tasks/my")
async def list_my_assigned_tasks(
    request: Request,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    status: str | None = Query(None),
    statuses: str | None = Query(
        None,
        description="Comma-separated statuses; takes precedence over `status`",
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Tasks assigned to the current user as a software owner (manual_notify).

    ``statuses`` (multi-value) lets the 我的待办 owner queue combine its
    actionable set in one round-trip — ``notified,failed`` so a FAILED
    provisioning job (the executor couldn't grant/revoke) surfaces next to
    the still-pending ones instead of hiding under an unfiltered "show all"
    the owner has to remember to toggle. A failed task is an action item:
    its ``error_message`` says why, and the owner can retry or hand it off.
    """
    statuses_list = (
        [s.strip() for s in (statuses or "").split(",") if s.strip()] or None
    )
    svc = ProvisioningService(db)
    tasks, total = await svc.list_tasks(
        page=page,
        page_size=page_size,
        status=status,
        statuses=statuses_list,
        assignee_id=current_user.id,
    )
    return paginated_response(
        data=tasks, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/tasks/{task_id}")
async def get_task(
    request: Request,
    task_id: UUID,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    task = await svc.get_task_response(task_id)
    return success_response(data=task, request=request)


@router.post("/tasks/{task_id}/complete")
@idempotent(
    "provisioning.complete", key_params=("task_id", "current_user.id"),
)
async def complete_task(
    request: Request,
    task_id: UUID,
    body: TaskCompleteRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Mark a manual_notify task as completed.

    Callable by either the task assignee (software owner) or an IT
    Admin. The router-level ``require_active_employee_for_mutations``
    dep already blocks leaver_pending callers; the per-row
    ``assignee_id == current_user.id`` check below is re-validated
    inside ``complete_manual_task`` after a SELECT FOR UPDATE so a
    concurrent admin reassignment can't race past it.
    """
    svc = ProvisioningService(db)
    task = await svc._get_task(task_id)

    is_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE)
    is_assignee = task.assignee_id == current_user.id
    if not (is_admin or is_assignee):
        raise ForbiddenError()

    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    updated = await svc.complete_manual_task(
        task_id,
        note=body.note,
        external_ref=body.external_ref,
        trace_id=trace_id,
        actor_user_id=current_user.id,
        actor_is_admin=is_admin,
    )
    await db.commit()
    return success_response(
        data=await svc.get_task_response(updated.id), request=request,
    )


# ── One-click no-login completion from the manual-notify link ───────
#
# These two routes are intentionally UNAUTHENTICATED — the signed
# action token in the URL is the authorisation (it was minted for one
# task + one recipient and expires). The router-level
# ``require_active_employee_for_mutations`` dep lets a credential-less
# POST through to the endpoint's own (absent) auth, and GET is always
# allowed, so no JWT is required here. See
# ``service._notify_manual_assignees`` for where the link is minted.

def _qc_page(
    *,
    title: str,
    heading: str,
    body_html: str,
    tone: str = "info",
    icon: str | None = None,
) -> HTMLResponse:
    """Render a tiny self-contained bilingual result page (no SPA, no
    login) for the one-click flow. Works in the Feishu in-app browser
    and any email client's external browser.

    ``tone`` (ok/err/warn/info) drives the accent colour + default badge
    icon shown in a circular header badge above the heading."""
    accents = {
        "ok": ("#1a7f37", "#e7f6ec", "✓"),
        "err": ("#cf222e", "#fdecec", "!"),
        "warn": ("#9a6700", "#fdf6e3", "!"),
        "info": ("#0969da", "#eaf2fb", "i"),
    }
    accent, badge_bg, default_icon = accents.get(tone, accents["info"])
    badge_glyph = icon if icon is not None else default_icon
    doc = (
        "<!doctype html><html lang='zh-CN'><head><meta charset='utf-8'>"
        "<meta name='viewport' content='width=device-width,initial-scale=1'>"
        "<meta name='robots' content='noindex'>"
        f"<title>{html.escape(title)}</title><style>"
        "*{box-sizing:border-box}"
        "body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',"
        "Roboto,Helvetica,Arial,'PingFang SC','Microsoft YaHei',sans-serif;"
        "background:linear-gradient(160deg,#f0f2f5 0%,#e8ebf0 100%);margin:0;"
        "min-height:100vh;display:flex;align-items:flex-start;"
        "justify-content:center;padding:24px;color:#1d1d1f}"
        ".wrap{width:100%;max-width:460px;margin-top:9vh;background:#fff;"
        "border-radius:20px;padding:40px 32px 28px;text-align:center;"
        "box-shadow:0 10px 40px rgba(0,0,0,.08),0 2px 8px rgba(0,0,0,.04)}"
        f".badge{{width:72px;height:72px;border-radius:50%;background:{badge_bg};"
        f"color:{accent};font-size:38px;font-weight:700;line-height:72px;"
        "margin:0 auto 22px;display:block}"
        "h1{font-size:22px;font-weight:600;margin:0 0 14px;letter-spacing:.3px;"
        f"color:{accent}}}"
        "p{line-height:1.65;color:#444;font-size:15px;margin:8px 0}"
        "p.en{color:#8a8a8e;font-size:13px}"
        ".btn{display:block;width:100%;border:0;background:#d4380d;color:#fff;"
        "font-size:16px;font-weight:600;padding:14px 22px;border-radius:12px;"
        "cursor:pointer;margin-top:22px;transition:opacity .15s;"
        "box-shadow:0 4px 14px rgba(212,56,13,.28)}"
        ".btn:active{opacity:.85}"
        ".lbl{display:block;text-align:left;font-size:13px;color:#6e6e73;"
        "margin:18px 0 6px;font-weight:600}"
        ".inp{display:block;width:100%;font-size:15px;padding:12px 14px;"
        "border:1px solid #d9d9de;border-radius:10px;background:#fafafa;"
        "color:#1d1d1f;outline:none}"
        ".inp:focus{border-color:#d4380d;background:#fff;"
        "box-shadow:0 0 0 3px rgba(212,56,13,.12)}"
        ".muted{color:#a1a1a6;font-size:12px;margin-top:26px;"
        "border-top:1px solid #f0f0f2;padding-top:18px}"
        "</style></head><body>"
        f"<div class='wrap'><span class='badge'>{badge_glyph}</span>"
        f"<h1>{heading}</h1>{body_html}"
        "<p class='muted'>Helix IT 平台 · 此链接由系统通知签发</p>"
        "</div></body></html>"
    )
    # Page-level CSP: the global SecurityHeadersMiddleware default
    # (``default-src 'none'``) would make the browser ignore the inline
    # <style> above and render this page as bare unstyled HTML. Allow
    # exactly what the page uses — inline styles + same-origin form
    # POST — and keep everything else locked down.
    return HTMLResponse(
        content=doc,
        headers={
            "Content-Security-Policy": (
                "default-src 'none'; style-src 'unsafe-inline'; "
                "form-action 'self'; frame-ancestors 'none'; "
                "base-uri 'none'"
            ),
        },
    )


def _qc_invalid_link() -> HTMLResponse:
    return _qc_page(
        title="链接无效",
        heading="链接无效或已过期",
        tone="err",
        body_html=(
            "<p>请在 Helix 控制台内处理该任务。</p>"
            "<p class='en'>This confirmation link is invalid or has expired.</p>"
        ),
    )


@router.get("/quick-complete", include_in_schema=False)
async def quick_complete_page(
    token: str = Query(...),
    db: AsyncSession = Depends(get_db),
):
    """Landing page for the one-click "我已处理,标记完成" link. GET has
    NO side effect (safe against email-client prefetch) — it just shows
    the task and a confirm button that POSTs back the same token."""
    try:
        payload = decode_action_token(
            token, expected_purpose=ACTION_TOKEN_PURPOSE_MANUAL_COMPLETE,
        )
        res_id = UUID(payload["res"])
    except (jwt.InvalidTokenError, ValueError, KeyError):
        return _qc_invalid_link()
    svc = ProvisioningService(db)
    try:
        task = await svc._get_task(res_id)
    except AppException:
        # Valid token but the task was hard-deleted / never existed —
        # show a friendly page, not a 500.
        return _qc_invalid_link()
    label = "回收" if task.action == "revoke" else "分配"
    if task.status in {"completed", "cancelled"}:
        return _qc_page(
            title="任务已完成",
            heading="该任务已完成",
            tone="ok",
            body_html=(
                (
                    # Same reason as the confirm page below: "hardware 回收"
                    # is not what this admin was asked to do.
                    "<p>该设备的 MDM 清理待办"
                    f"当前状态:{html.escape(task.status)},无需再操作。</p>"
                    if task.source == SOURCE_ASSET_DISPOSAL else
                    f"<p>任务「{html.escape(task.target_type)} {label}」"
                    f"当前状态:{html.escape(task.status)},无需再操作。</p>"
                )
                + "<p class='en'>This task is already resolved — nothing "
                  "to do.</p>"
            ),
        )
    if task.execution_mode != EXEC_MANUAL_NOTIFY:
        return _qc_page(
            title="无需手动",
            heading="该任务由系统自动处理",
            tone="info",
            body_html=(
                "<p>该任务由系统自动处理,无需手动操作。</p>"
                "<p class='en'>This task is auto-handled; no manual action "
                "needed.</p>"
            ),
        )
    safe_token = html.escape(token)

    def form_html(button: str) -> str:
        # A function, not a `.format()` template: the string carries the
        # token, and running `.format()` over attacker-influenced text is
        # a footgun waiting for the first token alphabet that contains a
        # brace. Nothing here is escaped twice — `safe_token` already is.
        return (
            "<form method='post' action='/api/v1/provisioning/quick-complete'>"
            f"<input type='hidden' name='token' value='{safe_token}'>"
            f"<button class='btn' type='submit'>{button}</button>"
            "</form>"
        )
    if task.source == SOURCE_ASSET_DISPOSAL:
        # An MDM device cleanup, not a seat revoke. The generic copy below
        # calls the subject a 坐席/账号 and promises to 「推进离职流程」 —
        # both are wrong here, and the second one states, at the exact
        # moment the admin confirms, the belief this whole feature exists
        # to remove: that MDM enrolment is something a departure triggers.
        # The device left company ownership; nobody has to have left.
        return _qc_page(
            title="确认完成",
            heading="确认已从 MDM 移除该设备?",
            tone="warn",
            icon="?",
            body_html=(
                "<p>点击下方按钮确认你<b>已在 MDM 控制台删除该设备记录</b>。"
                "确认后 Helix 会关闭这条待办 —— 下一次设备同步若仍能查到"
                "该设备,会重新提醒你。</p>"
                "<p class='en'>Click to confirm you have already removed "
                "the device record in the MDM console. Helix closes this "
                "todo; if the next device sync still finds the device, it "
                "will ask again.</p>"
                + form_html("✅ 确认已移除,标记完成")
            ),
        )
    if task.target_type == TARGET_SSO_GRANT:
        # SSO removal (migration 603): the click is NOT accepted as proof.
        # The server refuses a bare submission (REVOKE_PROOF_REQUIRED), so
        # the page asks for the reference up front instead of bouncing the
        # owner after they pressed the button.
        ctx = task.context_data if isinstance(task.context_data, dict) else {}
        system_name = html.escape(str(ctx.get("system_name") or "SSO"))
        provider = html.escape(str(ctx.get("sso_provider") or "IdP"))
        evidence_form = (
            "<form method='post' action='/api/v1/provisioning/quick-complete'>"
            f"<input type='hidden' name='token' value='{safe_token}'>"
            "<label class='lbl' for='external_ref'>凭证 / Evidence</label>"
            "<input class='inp' id='external_ref' name='external_ref' "
            "type='text' required minlength='3' maxlength='512' "
            "placeholder='工单号 / 控制台链接 / 一句说明 · ticket id, console URL or a note'>"
            "<button class='btn' type='submit'>✅ 已停用,提交凭证 · Disabled — submit</button>"
            "</form>"
        )
        return _qc_page(
            title="确认已停用 SSO",
            heading=f"确认已在 {provider} 停用 {system_name} 的 SSO 访问?",
            tone="warn",
            icon="?",
            body_html=(
                f"<p>请先在 <b>{provider} / 目标系统控制台</b>停用该账号对 "
                f"<b>{system_name}</b> 的 SSO 访问,再在下方填写凭证提交。"
                "没有凭证的确认不会被接受;Helix 之后会用登录审计复核这次停用,"
                "若撤销后仍有登录会重新打开任务。</p>"
                "<p class='en'>Disable this account's SSO access to "
                f"{system_name} in the {provider} / target console first, then "
                "record the evidence below. A confirmation without evidence is "
                "refused; Helix later verifies against the login audit and "
                "re-opens the task if a login shows up.</p>"
                + evidence_form
            ),
        )
    return _qc_page(
        title="确认完成",
        heading=f"确认已{label}该坐席/账号?",
        tone="warn",
        icon="?",
        body_html=(
            f"<p>点击下方按钮确认你<b>已在外部系统中完成{label}</b>。"
            "确认后 Helix 会关闭该任务并推进离职流程。</p>"
            "<p class='en'>Click to confirm you have already performed this "
            f"{html.escape(task.action)} in the external system.</p>"
            + form_html(f"✅ 确认已{label},标记完成")
        ),
    )


@router.post("/quick-complete", include_in_schema=False)
async def quick_complete_submit(
    request: Request,
    token: str = Form(...),
    external_ref: str | None = Form(None, max_length=512),
    db: AsyncSession = Depends(get_db),
):
    """Perform the one-click completion. The signed token authorises a
    single task for a single recipient — no Helix login required.

    ``external_ref`` is optional on the wire and REQUIRED by the service for
    evidence-gated targets (SSO removal, migration 603): the GET page
    renders the field for those tasks and the service refuses a bare
    submission with ``REVOKE_PROOF_REQUIRED``, which lands on the friendly
    error page below rather than silently closing an unverified removal."""
    try:
        payload = decode_action_token(
            token, expected_purpose=ACTION_TOKEN_PURPOSE_MANUAL_COMPLETE,
        )
        res_id = UUID(payload["res"])
        sub_id = UUID(payload["sub"])
        # A departed / force-logged-out recipient's stale 14-day link must not
        # still complete tasks — raises ValueError → friendly invalid-link page.
        await enforce_action_token_subject(db, payload)
    except (jwt.InvalidTokenError, ValueError, KeyError):
        return _qc_invalid_link()
    trace_id = str(getattr(request.state, "trace_id", "") or "")
    svc = ProvisioningService(db)
    try:
        await svc.complete_manual_task(
            res_id,
            note=None,
            external_ref=(external_ref or "").strip() or None,
            trace_id=trace_id,
            actor_user_id=sub_id,
            via_token=True,
        )
        await db.commit()
    except AppException as exc:
        await db.rollback()
        return _qc_page(
            title="无法完成",
            heading="无法标记完成",
            tone="err",
            body_html=f"<p>{html.escape(exc.message or exc.code)}</p>",
        )
    except Exception:
        # Non-AppException (IntegrityError / deadlock / KeyError in the
        # post-success side-effects) would otherwise escape as a raw 500 in
        # the recipient's Feishu/email in-app browser AND skip the rollback,
        # leaving the task mid-state. Roll back and show the same friendly
        # page as the GET path, matching this flow's "no 500" contract.
        await db.rollback()
        logger.exception(
            "quick_complete_submit failed for task res=%s trace=%s",
            res_id, trace_id,
        )
        return _qc_page(
            title="出错了",
            heading="处理时发生错误",
            tone="err",
            body_html=(
                "<p>系统处理时发生意外错误,请稍后重试;若持续失败请联系 IT。</p>"
                "<p class='en'>An unexpected error occurred — please retry, "
                "or contact IT if it persists.</p>"
            ),
        )
    return _qc_page(
        title="已完成",
        heading="已标记完成",
        tone="ok",
        body_html=(
            "<p>任务已关闭,离职流程会自动推进。感谢处理!</p>"
            "<p class='en'>Done — the task is closed and offboarding will "
            "proceed.</p>"
        ),
    )


@router.post("/tasks/{task_id}/fail")
@idempotent("provisioning.fail", key_params=("task_id", "current_user.id"))
async def fail_task(
    request: Request,
    task_id: UUID,
    body: TaskFailRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    updated = await svc.fail_task(
        task_id, reason=body.reason, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(
        data=await svc.get_task_response(updated.id), request=request,
    )


@router.post("/tasks/{task_id}/cancel")
@idempotent(
    "provisioning.cancel", key_params=("task_id", "current_user.id"),
)
async def cancel_task(
    request: Request,
    task_id: UUID,
    body: TaskFailRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    updated = await svc.cancel_task(
        task_id, reason=body.reason, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(
        data=await svc.get_task_response(updated.id), request=request,
    )


@router.post("/tasks/{task_id}/execute")
async def execute_task(
    request: Request,
    task_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Dispatch a single pending task. For auto_api tasks, calls the connector
    and records the result. For manual_notify, puts the task in 'notified'."""
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    task = await svc._get_task(task_id)
    await svc.execute_task(task, trace_id=trace_id, actor_user_id=current_user.id)
    await db.commit()
    return success_response(
        data=await svc.get_task_response(task.id), request=request,
    )


@router.post("/tasks/execute-bulk")
async def bulk_execute(
    request: Request,
    body: BulkExecuteRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Execute many pending tasks in one call. Used by the onboarding
    confirmation flow ('run all pending for this new hire') and by the
    admin 'run pending' bulk button.

    Refuses requests that mix ``task_ids`` with ``source`` /
    ``user_id`` filters. ``task_ids`` is a precise enumeration; the
    filters are a query. Mixing them silently drops the filters
    (``task_ids`` wins inside the service) which lets a
    social-engineered admin paste an attacker-supplied id list while
    believing they were also constrained by user_id. One or the
    other, never both.
    """
    if body.task_ids and (body.source or body.user_id):
        raise AppException(
            code="BULK_EXECUTE_AMBIGUOUS",
            message=(
                "task_ids cannot be combined with source / user_id "
                "filters. Pass either an explicit task_ids list OR a "
                "filter, not both."
            ),
            status_code=400,
        )
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    result = await svc.bulk_execute(body, trace_id=trace_id, actor_user_id=current_user.id)
    await db.commit()
    return success_response(data=result, request=request)


# ─── Provision policies (admin CRUD) ───


@router.get("/policies")
async def list_policies(
    request: Request,
    department_name: str | None = None,
    active_only: bool = False,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    policies = await svc.list_policies(
        department_name=department_name, active_only=active_only,
    )
    return success_response(data=policies, request=request)


@router.post("/policies")
async def create_policy(
    request: Request,
    body: ProvisionPolicyCreate,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    policy = await svc.create_policy(
        body, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=policy, request=request)


@router.patch("/policies/{policy_id}")
async def update_policy(
    request: Request,
    policy_id: UUID,
    body: ProvisionPolicyUpdate,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    policy = await svc.update_policy(
        policy_id, body, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data=policy, request=request)


@router.delete("/policies/{policy_id}")
async def delete_policy(
    request: Request,
    policy_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    await svc.delete_policy(policy_id, trace_id=trace_id, actor_user_id=current_user.id)
    await db.commit()
    return success_response(data=None, request=request)


# ─── Onboarding orchestration ───


@router.get("/onboarding/preview/{person_id}")
async def preview_onboarding(
    request: Request,
    person_id: UUID,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Show what would be provisioned for a person given current policies,
    marking items already satisfied."""
    svc = ProvisioningService(db)
    preview = await svc.preview_onboarding(person_id)
    return success_response(data=preview, request=request)


@router.get("/onboarding-progress")
async def onboarding_progress(
    request: Request,
    days: int = 90,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Per-hire onboarding progress summary.

    Replaces the frontend's N+1 query pattern (``OnboardingProgressPage``
    previously called ``listTasks`` once per person). Returns one row
    per hire in the last ``days`` days with aggregated task counts:

        [
          {
            user_id, display_name, hire_date,
            total, completed, pending, failed, notified,
          },
          ...
        ]

    Plus a top-level summary: {total_hires, hires_with_tasks,
    active_onboardings, fully_done}.

    Uses two queries total (hires + tasks-grouped-by-user-and-status)
    instead of N+1. Backend-compute is the CLAUDE.md "Frontend is pure
    presentation" rule in action — the page now renders what the
    server tells it to.
    """
    from datetime import date as _date
    from datetime import timedelta as _td

    from sqlalchemy import and_, func, select

    from app.domains.provisioning.models import SOURCE_ONBOARDING, ProvisioningTask
    from app.identity.person_models import IdentityPerson

    today = _date.today()
    # Clamp days to [1, 365]. Admin-only endpoint, but an unbounded scan
    # (days=99999) would touch every employee + every onboarding task ever.
    cutoff = today - _td(days=max(1, min(days, 365)))

    hire_rows = await db.execute(
        select(IdentityPerson)
        .where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.hire_date.isnot(None),
            IdentityPerson.hire_date >= cutoff,
        )
        .order_by(IdentityPerson.hire_date.desc()),
    )
    hires = list(hire_rows.scalars().all())
    if not hires:
        return success_response(
            data={
                "hires": [],
                "summary": {
                    "total_hires": 0,
                    "hires_with_tasks": 0,
                    "active_onboardings": 0,
                    "fully_done": 0,
                },
            },
            request=request,
        )

    # ONE grouped query for task counts across all hires — instead of
    # one listTasks call per person like the frontend used to do.
    hire_ids = [p.id for p in hires]
    task_agg = await db.execute(
        select(
            ProvisioningTask.user_id,
            ProvisioningTask.status,
            func.count(ProvisioningTask.id).label("n"),
        )
        .where(
            and_(
                ProvisioningTask.user_id.in_(hire_ids),
                ProvisioningTask.source == SOURCE_ONBOARDING,
            ),
        )
        .group_by(ProvisioningTask.user_id, ProvisioningTask.status),
    )
    # user_id → {status → count}
    counts: dict[str, dict[str, int]] = {}
    for uid, status, n in task_agg.all():
        counts.setdefault(str(uid), {})[status] = int(n or 0)

    rows = []
    total_hires = len(hires)
    hires_with_tasks = 0
    active_onboardings = 0
    fully_done = 0
    for p in hires:
        c = counts.get(str(p.id), {})
        completed = c.get("completed", 0)
        failed = c.get("failed", 0)
        notified = c.get("notified", 0)
        pending = c.get("pending", 0) + c.get("in_progress", 0)
        total = completed + failed + notified + pending
        rows.append({
            "user_id": str(p.id),
            "display_name": p.display_name,
            "work_email": p.work_email,
            "hire_date": p.hire_date.isoformat() if p.hire_date else None,
            "total": total,
            "completed": completed,
            "pending": pending,
            "failed": failed,
            "notified": notified,
        })
        if total > 0:
            hires_with_tasks += 1
            if pending + notified > 0:
                active_onboardings += 1
            elif failed == 0:
                fully_done += 1

    return success_response(
        data={
            "hires": rows,
            "summary": {
                "total_hires": total_hires,
                "hires_with_tasks": hires_with_tasks,
                "active_onboardings": active_onboardings,
                "fully_done": fully_done,
            },
        },
        request=request,
    )


@router.get("/role-change/preview/{person_id}")
async def preview_role_change(
    request: Request,
    person_id: UUID,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Compute policy diff for a person (add / remove items)."""
    svc = ProvisioningService(db)
    preview = await svc.preview_role_change(person_id)
    return success_response(data=preview, request=request)


@router.post("/role-change/apply/{person_id}")
async def apply_role_change(
    request: Request,
    person_id: UUID,
    body: ReconcileApplyRequest,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    count = await svc.apply_role_change(
        person_id, body, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(data={"tasks_created": count}, request=request)


@router.get("/billing")
async def compute_billing(
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    """Current monthly cost aggregated from active seat allocations."""
    svc = ProvisioningService(db)
    result = await svc.compute_billing()
    return success_response(data=result, request=request)


@router.get("/billing/my")
async def compute_billing_my(
    request: Request,
    scope: str = Query(
        "me",
        pattern=r"^(me|subordinates)$",
        description=(
            "me → only the caller's own active seats. "
            "subordinates → caller + every direct and transitive report "
            "in ``identity_persons.manager_person_id``."
        ),
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Employee self-service billing view.

    ``scope=me`` mirrors the line items on ``/me`` for just the caller;
    ``scope=subordinates`` adds every direct/transitive report so a
    people-manager can see their team's software spend without having
    ``asset_admin`` on the admin billing page.
    """
    user_ids: list[UUID] = [current_user.id]
    if scope == "subordinates":
        # Self ∪ every direct/transitive report (excluding ``left``) — the
        # ``MANAGED`` scope of the canonical resolver. Previously an inline
        # BFS here applied ``employment_status != 'left'`` but, unlike the
        # canonical helper, omitted the ``deleted_at IS NULL`` guard, so a
        # soft-deleted report could leak into a manager's billing scope.
        subs = await subordinate_person_ids(
            db, current_user.id, scope=SubordinateScope.MANAGED,
        )
        user_ids = [current_user.id, *subs]

    svc = ProvisioningService(db)
    result = await svc.compute_billing(user_ids=user_ids)
    return success_response(
        data=result,
        request=request,
        meta={"scope": scope, "user_count": len(user_ids)},
    )


@router.post("/onboarding/execute/{person_id}")
async def execute_onboarding(
    request: Request,
    person_id: UUID,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Create pending provisioning tasks for the person (idempotent)."""
    svc = ProvisioningService(db)
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    tasks = await svc.execute_onboarding_for_person(
        person_id, trace_id=trace_id, actor_user_id=current_user.id,
    )
    await db.commit()
    return success_response(
        data={"task_count": len(tasks), "task_ids": [str(t.id) for t in tasks]},
        request=request,
    )


@router.post("/tasks/{task_id}/retry")
# Retry re-enqueues the driver call; a replayed retry would enqueue a
# second attempt for the same task.
@idempotent("provisioning.retry", key_params=("task_id", "current_user.id"))
async def retry_task(
    request: Request,
    task_id: UUID,
    body: TaskRetryRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Re-run a FAILED provisioning task.

    Callable by the task assignee (software owner) or an IT Admin — mirrors
    ``complete_task`` so the 我的待办 owner queue's 重试 button works for the
    non-admin owners it now surfaces failed jobs to (结果闭环 PR3 review
    fix; previously ``require_admin()`` 403'd them into a dead-end). Only an
    admin may ``reset_assignee`` (re-deriving the assignee is an admin op);
    an assignee retry keeps the task on themselves. Assignee re-validated
    under SELECT FOR UPDATE inside ``retry_task``.
    """
    svc = ProvisioningService(db)
    task = await svc._get_task(task_id)
    is_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE)
    is_assignee = task.assignee_id == current_user.id
    if not (is_admin or is_assignee):
        raise ForbiddenError()
    trace_id = str(request.state.trace_id) if hasattr(request.state, "trace_id") else ""
    updated = await svc.retry_task(
        task_id,
        reset_assignee=body.reset_assignee and is_admin,
        trace_id=trace_id,
        actor_user_id=current_user.id,
        actor_is_admin=is_admin,
    )
    await db.commit()
    return success_response(
        data=await svc.get_task_response(updated.id), request=request,
    )


# ─── Provisioning bundles (PR-2 of symmetric-onboarding, 2026-06-01) ───
#
# Bundles group N ``ProvisionPolicy`` rows under a named kit. Admin
# applies a bundle to a single onboarding process via the cockpit
# detail page (`POST /bundles/{id}/apply-to-person`). The matching
# engine is unchanged — bundles are a grouping + apply surface only.


def _serialize_bundle(
    bundle, member_count: int | None = None,
) -> dict:
    out = {
        "id": str(bundle.id),
        "name": bundle.name,
        "description": bundle.description,
        "target_role_hint": bundle.target_role_hint,
        "is_active": bool(bundle.is_active),
        "created_by_person_id": (
            str(bundle.created_by_person_id)
            if bundle.created_by_person_id
            else None
        ),
        "created_at": (
            bundle.created_at.isoformat() if bundle.created_at else None
        ),
        "updated_at": (
            bundle.updated_at.isoformat() if bundle.updated_at else None
        ),
    }
    if member_count is not None:
        out["member_count"] = int(member_count)
    return out


def _serialize_policy(policy) -> dict:
    return {
        "id": str(policy.id),
        "department_name": policy.department_name,
        "match_job_title": policy.match_job_title,
        "match_location": policy.match_location,
        "target_type": policy.target_type,
        "target_id": str(policy.target_id),
        "is_active": bool(policy.is_active),
        "bundle_id": str(policy.bundle_id) if policy.bundle_id else None,
    }


@router.get("/bundles")
async def list_bundles(
    request: Request,
    active_only: bool = Query(default=False),
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    rows = await BundleService(db).list_bundles(active_only=active_only)
    return success_response(
        data=[_serialize_bundle(b, c) for b, c in rows],
        request=request,
    )


@router.get("/bundles/{bundle_id}")
async def get_bundle(
    bundle_id: UUID,
    request: Request,
    _admin: AuthUser = Depends(require_lifecycle_read()),
    db: AsyncSession = Depends(get_db),
):
    bundle, policies = await BundleService(db).get_bundle(bundle_id)
    return success_response(
        data={
            "bundle": _serialize_bundle(bundle, len(policies)),
            "policies": [_serialize_policy(p) for p in policies],
        },
        request=request,
    )


class BundleCreateBody(BaseModel):
    name: str = Field(..., min_length=1, max_length=128)
    description: str | None = None
    target_role_hint: str | None = None


@router.post("/bundles", status_code=201)
async def create_bundle(
    body: BundleCreateBody,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    trace_id = getattr(request.state, "trace_id", body.name)
    bundle = await BundleService(db).create_bundle(
        name=body.name,
        description=body.description,
        target_role_hint=body.target_role_hint,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(data=_serialize_bundle(bundle, 0), request=request)


class BundleUpdateBody(BaseModel):
    description: str | None = None
    target_role_hint: str | None = None
    is_active: bool | None = None


@router.patch("/bundles/{bundle_id}")
async def update_bundle(
    bundle_id: UUID,
    body: BundleUpdateBody,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    trace_id = getattr(request.state, "trace_id", str(bundle_id))
    bundle = await BundleService(db).update_bundle(
        bundle_id,
        description=body.description,
        target_role_hint=body.target_role_hint,
        is_active=body.is_active,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(data=_serialize_bundle(bundle), request=request)


@router.delete("/bundles/{bundle_id}", status_code=204)
async def delete_bundle(
    bundle_id: UUID,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    trace_id = getattr(request.state, "trace_id", str(bundle_id))
    await BundleService(db).delete_bundle(
        bundle_id,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()


@router.post("/bundles/{bundle_id}/members/{policy_id}", status_code=201)
async def add_bundle_member(
    bundle_id: UUID,
    policy_id: UUID,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    trace_id = getattr(request.state, "trace_id", str(bundle_id))
    member = await BundleService(db).add_member(
        bundle_id,
        policy_id,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(
        data={
            "id": str(member.id),
            "bundle_id": str(member.bundle_id),
            "policy_id": str(member.policy_id),
        },
        request=request,
    )


@router.delete("/bundles/{bundle_id}/members/{policy_id}", status_code=204)
async def remove_bundle_member(
    bundle_id: UUID,
    policy_id: UUID,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    trace_id = getattr(request.state, "trace_id", str(bundle_id))
    await BundleService(db).remove_member(
        bundle_id,
        policy_id,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()


class ApplyBundleBody(BaseModel):
    person_id: UUID


@router.post("/bundles/{bundle_id}/apply-to-person")
async def apply_bundle_to_person(
    bundle_id: UUID,
    body: ApplyBundleBody,
    request: Request,
    admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    trace_id = getattr(request.state, "trace_id", str(bundle_id))
    result = await BundleService(db).apply_to_person(
        bundle_id,
        body.person_id,
        trace_id=trace_id,
        actor_user_id=admin.id,
    )
    await db.commit()
    return success_response(data=result, request=request)