"""Manager team-view aggregations.

Two endpoints feed off this module:

  * ``GET /api/v1/me/team/footprint`` — one row per direct + transitive
    report (bounded BFS) with cross-domain resource counts + last-month
    cloud cost. Drives ``/my/team/footprint``.

  * ``GET /api/v1/me/team/cloud`` — per-resource detail for cloud
    grants / scopes / DNS / CDN / domains where one of the manager's
    subordinates is the owner. Drives ``/my/team/cloud``.

RBAC: the caller must be an active employee. An empty subordinates
set returns 200 OK with empty rows so the UI degrades into a "no
team" empty state rather than 403'ing a leaf-node employee.

Schema quirks honoured here (see memory ``reference_asyncmy_uuid_bind``):
  * ``cmdb_*.owner_person_id`` are ``String(36)`` columns, so we
    MUST pass ``str(uuid)`` not ``uuid.UUID`` in the IN clauses or
    asyncmy silently returns 0 rows.
  * ``CmdbCloudAccessGrant.person_id`` / ``.status`` / ``.role_key``
    (not ``assignee_person_id`` / ``grant_status`` / ``role_name``).
  * ``ApiKeyAssignment.assigned_to`` (not ``assignee_person_id``).
  * ``CmdbDomain.domain_name`` / ``CmdbCdnDomain.domain_name`` /
    ``CmdbDnsRecord.record_type`` / ``CmdbCloudScope.display_name``.
"""
from __future__ import annotations

import uuid
from datetime import date, timedelta
from typing import Any

from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.person_scope import SubordinateScope, subordinate_person_ids
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
from app.domains.apikey.spend import classify_bucket, compute_assignment_spend
from app.domains.asset.models import AssetItem, SoftwareSeatAllocation
from app.domains.cmdb.cost_service import subordinates_cost_breakdown
from app.domains.cmdb.models import (
    CmdbCdnDomain,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbDnsRecord,
    CmdbDomain,
    owner_governed_scope_criteria,
)
from app.identity.person_models import IdentityPerson

# ── Helpers ──────────────────────────────────────────────────────


async def _list_subordinates(
    db: AsyncSession, manager_id: uuid.UUID, *, include_left: bool = False,
) -> list[IdentityPerson]:
    """Direct + transitive reports.

    Excludes ``employment_status == 'left'`` by default — a leaver keeps
    their ``manager_person_id`` pointer and a non-deleted row, so without
    this filter they'd be counted as a current report forever (same root
    cause fixed for the dashboard ``_subordinate_ids`` in PR #782). The
    manager can opt back in via ``include_left`` to review what a former
    report still holds (e.g. for re-assignment) — that maps to the
    canonical ``ALL`` scope; the default maps to ``MANAGED`` (``left``
    out, ``leaver_pending`` kept).
    """
    scope = SubordinateScope.ALL if include_left else SubordinateScope.MANAGED
    sub_ids = await subordinate_person_ids(db, manager_id, scope=scope)
    if not sub_ids:
        return []
    rows = (
        await db.execute(
            select(IdentityPerson)
            .where(
                IdentityPerson.id.in_(sub_ids),
                IdentityPerson.deleted_at.is_(None),
            )
            .order_by(IdentityPerson.display_name.asc())
        )
    ).scalars().all()
    return list(rows)


async def _counts_by_person(
    db: AsyncSession,
    table_col,
    person_id_col,
    person_ids: list[str],
    extra_where: list[Any] | None = None,
) -> dict[str, int]:
    if not person_ids:
        return {}
    clauses: list[Any] = [person_id_col.in_(person_ids)]
    if extra_where:
        clauses.extend(extra_where)
    rows = (
        await db.execute(
            select(person_id_col, func.count(table_col))
            .where(and_(*clauses))
            .group_by(person_id_col)
        )
    ).all()
    return {str(pid): int(n) for pid, n in rows if pid is not None}


# ── Public service methods ───────────────────────────────────────


async def build_team_footprint(
    db: AsyncSession,
    *,
    manager: IdentityPerson,
    include_left: bool = False,
) -> dict[str, Any]:
    """One row per direct/transitive report + per-domain counts.

    Cost is for the **previous calendar month** so the number is
    closed and stable (the current month moves throughout the day
    and would create spurious churn in the UI).

    Leavers (``employment_status == 'left'``) are hidden unless
    ``include_left`` is set — see ``_list_subordinates``.
    """
    subordinates = await _list_subordinates(db, manager.id, include_left=include_left)
    if not subordinates:
        return {
            "manager_id": str(manager.id),
            "subordinates_count": 0,
            "rows": [],
            "cost_window": None,
        }

    sub_ids_str = [str(p.id) for p in subordinates]

    # ── Asset (hardware) — owned by the person, currently allocated ─
    asset_counts = await _counts_by_person(
        db,
        AssetItem.id,
        AssetItem.assigned_to,
        sub_ids_str,
        extra_where=[AssetItem.assigned_to.isnot(None)],
    )

    # ── Software seat allocations (active) ─────────────────────
    seat_counts = await _counts_by_person(
        db,
        SoftwareSeatAllocation.id,
        SoftwareSeatAllocation.user_id,
        sub_ids_str,
        extra_where=[SoftwareSeatAllocation.status == "active"],
    )

    # ── API key assignments (active) ───────────────────────────
    api_key_counts = await _counts_by_person(
        db,
        ApiKeyAssignment.id,
        ApiKeyAssignment.assigned_to,
        sub_ids_str,
        extra_where=[ApiKeyAssignment.status == "active"],
    )

    # ── Cloud access grants (active) ───────────────────────────
    grant_counts = await _counts_by_person(
        db,
        CmdbCloudAccessGrant.id,
        CmdbCloudAccessGrant.person_id,
        sub_ids_str,
        extra_where=[CmdbCloudAccessGrant.status == "active"],
    )

    # ── Cloud scopes owned ─────────────────────────────────────
    # MUST use the same predicate as the scope LIST further down — this
    # is the per-report count badge next to it, and a badge that says 5
    # over a list of 3 is its own bug report.
    scope_counts = await _counts_by_person(
        db,
        CmdbCloudScope.id,
        CmdbCloudScope.owner_person_id,
        sub_ids_str,
        extra_where=owner_governed_scope_criteria(),
    )

    # ── DNS records owned ──────────────────────────────────────
    dns_counts = await _counts_by_person(
        db,
        CmdbDnsRecord.id,
        CmdbDnsRecord.owner_person_id,
        sub_ids_str,
        extra_where=[CmdbDnsRecord.deleted_at.is_(None)],
    )

    # ── CDN domains owned ──────────────────────────────────────
    cdn_counts = await _counts_by_person(
        db,
        CmdbCdnDomain.id,
        CmdbCdnDomain.owner_person_id,
        sub_ids_str,
        extra_where=[CmdbCdnDomain.deleted_at.is_(None)],
    )

    # ── Root domains owned ─────────────────────────────────────
    domain_counts = await _counts_by_person(
        db,
        CmdbDomain.id,
        CmdbDomain.owner_person_id,
        sub_ids_str,
        extra_where=[CmdbDomain.deleted_at.is_(None)],
    )

    # ── Last-month cloud cost (per person) ─────────────────────
    today = date.today()  # noqa: DTZ011 — calendar date, no TZ math
    first_of_this_month = today.replace(day=1)
    last_of_prev = first_of_this_month - timedelta(days=1)
    first_of_prev = last_of_prev.replace(day=1)
    cost_breakdown = await subordinates_cost_breakdown(
        db,
        person_ids={p.id for p in subordinates},
        start=first_of_prev,
        end=last_of_prev,
    )
    cost_by_person: dict[str, float] = {
        str(row["person_id"]): float(row["cost_usd"])
        for row in cost_breakdown.get("by_person", [])
        if row.get("person_id")
    }

    # ── Stitch into rows ────────────────────────────────────────
    rows: list[dict[str, Any]] = []
    for p in subordinates:
        pid = str(p.id)
        rows.append({
            "person_id": pid,
            "display_name": p.display_name,
            "work_email": p.work_email,
            # ``department`` is a Python @property derived from
            # primary_department/department_chain — safe to read.
            "department": p.department,
            "employment_status": p.employment_status,
            "manager_person_id": (
                str(p.manager_person_id)
                if p.manager_person_id else None
            ),
            "footprint": {
                "assets": asset_counts.get(pid, 0),
                "seats": seat_counts.get(pid, 0),
                "api_keys": api_key_counts.get(pid, 0),
                "cloud_grants": grant_counts.get(pid, 0),
                "scopes_owned": scope_counts.get(pid, 0),
                "dns_owned": dns_counts.get(pid, 0),
                "cdn_owned": cdn_counts.get(pid, 0),
                "domains_owned": domain_counts.get(pid, 0),
                "cost_last_month_usd": round(cost_by_person.get(pid, 0.0), 2),
            },
        })

    # Sort by total resource footprint desc so the heaviest report
    # floats to the top — that's the operator's main signal for
    # "who needs review."
    def _row_score(r: dict[str, Any]) -> int:
        f = r["footprint"]
        return (
            f["assets"]
            + f["seats"]
            + f["api_keys"]
            + f["cloud_grants"]
            + f["scopes_owned"]
            + f["dns_owned"]
            + f["cdn_owned"]
            + f["domains_owned"]
        )

    rows.sort(key=_row_score, reverse=True)

    return {
        "manager_id": str(manager.id),
        "subordinates_count": len(subordinates),
        "rows": rows,
        "cost_window": {
            "start": first_of_prev.isoformat(),
            "end": last_of_prev.isoformat(),
        },
    }


async def build_team_cloud(
    db: AsyncSession,
    *,
    manager: IdentityPerson,
    include_left: bool = False,
) -> dict[str, Any]:
    """Cloud-domain detail across direct/transitive reports.

    Five lists (grants / scopes / dns / cdn / domains) — one row per
    resource, enriched with the owner's display name so the manager
    can scan "who owns what" without a second fetch.

    Leavers (``employment_status == 'left'``) are hidden unless
    ``include_left`` is set — see ``_list_subordinates``.
    """
    subordinates = await _list_subordinates(db, manager.id, include_left=include_left)
    if not subordinates:
        return {
            "manager_id": str(manager.id),
            "subordinates_count": 0,
            "grants": [],
            "scopes": [],
            "dns": [],
            "cdn": [],
            "domains": [],
        }

    sub_ids_str = [str(p.id) for p in subordinates]
    by_id = {str(p.id): p for p in subordinates}

    def _owner_pack(pid: str | None) -> dict[str, str | None]:
        if not pid:
            return {"id": None, "name": None, "email": None}
        person = by_id.get(str(pid))
        if person is None:
            return {"id": str(pid), "name": None, "email": None}
        return {
            "id": str(person.id),
            "name": person.display_name,
            "email": person.work_email,
        }

    # ── Active cloud grants — assignee in subordinates ─────────
    grant_rows = (
        await db.execute(
            select(CmdbCloudAccessGrant)
            .where(
                CmdbCloudAccessGrant.person_id.in_(sub_ids_str),
                CmdbCloudAccessGrant.status == "active",
            )
            .order_by(CmdbCloudAccessGrant.granted_at.desc())
            .limit(500)
        )
    ).scalars().all()
    grants_out = [{
        "id": str(g.id),
        "scope_id": str(g.scope_id) if g.scope_id else None,
        "role_key": g.role_key,
        "role_display_name": g.role_display_name,
        "status": g.status,
        "expires_at": g.expires_at.isoformat() if g.expires_at else None,
        "assignee": _owner_pack(g.person_id),
        "granted_at": g.granted_at.isoformat() if g.granted_at else None,
    } for g in grant_rows]

    # ── Scopes where one of the subs is owner ─────────────────
    # Same predicate as offboarding / mover holdings: this list is what
    # a manager is told their reports are responsible for, so it must
    # not include decommissioned accounts or cost-allocation rows the
    # admin console itself hides (2026-09-07 audit — see
    # ``owner_governed_scope_criteria``).
    scope_rows = (
        await db.execute(
            select(CmdbCloudScope)
            .where(
                CmdbCloudScope.owner_person_id.in_(sub_ids_str),
                *owner_governed_scope_criteria(),
            )
            .order_by(CmdbCloudScope.display_name.asc())
            .limit(500)
        )
    ).scalars().all()
    scopes_out = [{
        "id": str(s.id),
        "display_name": s.display_name,
        "scope_type": s.scope_type,
        "external_id": s.external_id,
        "status": s.status,
        "environment": s.environment,
        "owner": _owner_pack(s.owner_person_id),
    } for s in scope_rows]

    # ── DNS records owned by subs ─────────────────────────────
    dns_rows = (
        await db.execute(
            select(CmdbDnsRecord)
            .where(
                CmdbDnsRecord.owner_person_id.in_(sub_ids_str),
                CmdbDnsRecord.deleted_at.is_(None),
            )
            .order_by(CmdbDnsRecord.name.asc())
            .limit(500)
        )
    ).scalars().all()
    dns_out = [{
        "id": str(r.id),
        "zone_name": r.zone_name,
        "name": r.name,
        "record_type": r.record_type,
        "owner": _owner_pack(r.owner_person_id),
    } for r in dns_rows]

    # ── CDN domains owned ──────────────────────────────────────
    cdn_rows = (
        await db.execute(
            select(CmdbCdnDomain)
            .where(
                CmdbCdnDomain.owner_person_id.in_(sub_ids_str),
                CmdbCdnDomain.deleted_at.is_(None),
            )
            .order_by(CmdbCdnDomain.domain_name.asc())
            .limit(500)
        )
    ).scalars().all()
    cdn_out = [{
        "id": str(c.id),
        "domain_name": c.domain_name,
        "status": c.status,
        "owner": _owner_pack(c.owner_person_id),
    } for c in cdn_rows]

    # ── Root domains owned ─────────────────────────────────────
    domain_rows = (
        await db.execute(
            select(CmdbDomain)
            .where(
                CmdbDomain.owner_person_id.in_(sub_ids_str),
                CmdbDomain.deleted_at.is_(None),
            )
            .order_by(CmdbDomain.domain_name.asc())
            .limit(500)
        )
    ).scalars().all()
    domains_out = [{
        "id": str(d.id),
        "domain_name": d.domain_name,
        "owner": _owner_pack(d.owner_person_id),
    } for d in domain_rows]

    return {
        "manager_id": str(manager.id),
        "subordinates_count": len(subordinates),
        "grants": grants_out,
        "scopes": scopes_out,
        "dns": dns_out,
        "cdn": cdn_out,
        "domains": domains_out,
    }


# ── Team API keys (held AI keys + per-key month-to-date spend) ──────

# A held key is one whose secret is live: ``active`` (issued) or
# ``pending_release`` (release requested but not yet revoked — the key
# still works and still bills). ``pending_approval`` has no key value
# yet and ``revoked`` is dead, so both are excluded.
_HELD_KEY_STATUSES = ("active", "pending_release")


def _current_month_window() -> tuple[date, date, str]:
    """Current month [1st, today] anchored on Asia/Shanghai.

    Mirrors :meth:`ApiKeyService.get_assignment_spend`'s window so the
    team roll-up, the per-AK detail panel, and the budget-alert cron all
    see the same month boundary (``month_end`` clamped to today_shanghai,
    never the future end-of-month).
    """
    from datetime import datetime
    from zoneinfo import ZoneInfo

    today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    month_start = today_sh.replace(day=1)
    return month_start, today_sh, f"{today_sh.year:04d}-{today_sh.month:02d}"


async def build_team_api_keys(
    db: AsyncSession,
    *,
    manager: IdentityPerson,
    include_left: bool = False,
) -> dict[str, Any]:
    """AI API keys held by direct/transitive reports + per-key MTD spend.

    Closes the manager-scope gap the team-cloud page left open: a lead
    could see a report's cloud grants / scopes / DNS but had no view of
    which AI API keys (OpenAI / Bedrock / Vertex / Bailian …) the team
    holds or what they cost this month. Each row carries the same
    month-to-date spend the per-AK detail page shows — computed through
    :func:`compute_assignment_spend`, so the numbers reconcile to the
    cent across surfaces.

    Read-only. The manager already sees a report's last-month cloud cost
    on the footprint page, so per-key spend visibility introduces no new
    authority. Mutations (apply / revoke / budget edit) stay on the
    employee's own ``/me/api-keys`` surface and the admin console.

    Leavers (``employment_status == 'left'``) are hidden unless
    ``include_left`` is set — same scope rule as the other team views.
    """
    subordinates = await _list_subordinates(db, manager.id, include_left=include_left)
    month_start, month_end, month_label = _current_month_window()
    empty_summary = {
        "count": 0,
        "total_mtd_usd": 0.0,
        "total_budget_usd": 0.0,
        "over_budget_count": 0,
        "approaching_count": 0,
        "data_unavailable_count": 0,
    }
    if not subordinates:
        return {
            "manager_id": str(manager.id),
            "subordinates_count": 0,
            "month": month_label,
            "api_keys": [],
            "summary": empty_summary,
        }

    sub_ids_str = [str(p.id) for p in subordinates]
    by_id = {str(p.id): p for p in subordinates}

    def _owner_pack(pid: Any) -> dict[str, str | None]:
        person = by_id.get(str(pid)) if pid else None
        if person is None:
            return {"id": str(pid) if pid else None, "name": None, "email": None}
        return {
            "id": str(person.id),
            "name": person.display_name,
            "email": person.work_email,
        }

    assignments = (
        await db.execute(
            select(ApiKeyAssignment)
            .where(
                ApiKeyAssignment.assigned_to.in_(sub_ids_str),
                ApiKeyAssignment.status.in_(_HELD_KEY_STATUSES),
            )
            .order_by(ApiKeyAssignment.created_at.desc())
            .limit(500)
        )
    ).scalars().all()

    # Provider display names in one round-trip (avoid N lazy-loads).
    provider_names: dict[str, str] = {}
    if assignments:
        prov_ids = {a.provider_id for a in assignments}
        provider_names = {
            str(pid): name
            for pid, name in (
                await db.execute(
                    select(ApiKeyProvider.id, ApiKeyProvider.name).where(
                        ApiKeyProvider.id.in_(prov_ids),
                    )
                )
            ).all()
        }

    rows: list[dict[str, Any]] = []
    total_mtd = 0.0
    total_budget = 0.0
    over_budget_count = 0
    approaching_count = 0
    data_unavailable_count = 0

    for a in assignments:
        provider_name = provider_names.get(str(a.provider_id))
        spend = await compute_assignment_spend(
            db, assignment=a, month_start=month_start, month_end=month_end,
        )
        data_available = bool(spend["data_available"])
        mtd_usd = round(float(spend["mtd_usd"] or 0), 2)
        budget_usd = float(a.monthly_budget_usd or 0)
        total_budget += budget_usd
        utilization_pct: float | None = None
        over_budget = False
        approaching = False
        if data_available:
            total_mtd += mtd_usd
            if budget_usd > 0:
                utilization_pct = round(mtd_usd / budget_usd * 100.0, 1)
                if mtd_usd > budget_usd:
                    over_budget = True
                elif utilization_pct >= 80.0:
                    approaching = True
        else:
            data_unavailable_count += 1
        if over_budget:
            over_budget_count += 1
        elif approaching:
            approaching_count += 1

        rows.append({
            "id": str(a.id),
            "owner": _owner_pack(a.assigned_to),
            "provider_name": provider_name,
            "bucket": spend["bucket"] or classify_bucket(provider_name),
            "project_name": a.project_name,
            "status": a.status,
            "valid_until": a.valid_until.isoformat() if a.valid_until else None,
            "monthly_budget_usd": round(budget_usd, 2),
            "mtd_usd": mtd_usd,
            "utilization_pct": utilization_pct,
            "over_budget": over_budget,
            "data_available": data_available,
            "data_caveat": spend["data_caveat"],
            "last_fact_date": spend["last_fact_date"],
        })

    # Heaviest spender first; keys with unresolved spend sink to the
    # bottom so the actionable rows (real $) lead.
    rows.sort(key=lambda r: (r["data_available"], r["mtd_usd"]), reverse=True)

    return {
        "manager_id": str(manager.id),
        "subordinates_count": len(subordinates),
        "month": month_label,
        "api_keys": rows,
        "summary": {
            "count": len(rows),
            "total_mtd_usd": round(total_mtd, 2),
            "total_budget_usd": round(total_budget, 2),
            "over_budget_count": over_budget_count,
            "approaching_count": approaching_count,
            "data_unavailable_count": data_unavailable_count,
        },
    }


# ── Team lifecycle (joiners onboarding-readiness + leavers offboarding) ──

_TERMINAL_PROC = ("completed", "cancelled")


def _today() -> date:
    from datetime import UTC, datetime
    return datetime.now(UTC).date()


async def build_team_lifecycle(
    db: AsyncSession,
    *,
    manager: IdentityPerson,
) -> dict[str, Any]:
    """Per-subordinate people-lifecycle cockpit for a manager.

    Closes the manager-scope parity gap: a manager could already see a
    report's assets / seats / cloud / cost / requests, but had NO view of
    where a report sits in onboarding (Day-1 readiness) or offboarding
    (exit countdown + blockers + asset-return status). Read-only — every
    actionable affordance deep-links to the manager's existing surfaces
    (device-inheritance preference on /my/team/joiners, etc.). Zero new
    push surface.

    Scope is the manager's direct + transitive reports (same BFS as
    footprint/cloud), ``include_left=True`` so leavers still in an open
    offboarding process surface. Only people with an OPEN onboarding or
    offboarding process produce a row — long-settled actives are omitted.
    """
    from app.domains.offboarding.models import (
        OffboardingBlocker,
        OffboardingProcess,
    )
    from app.domains.onboarding.models import (
        OnboardingBlocker,
        OnboardingProcess,
    )

    subordinates = await _list_subordinates(db, manager.id, include_left=True)
    if not subordinates:
        return {
            "manager_id": str(manager.id),
            "subordinates_count": 0,
            "joiners": [],
            "leavers": [],
            "summary": {
                "joiners_count": 0, "leavers_count": 0,
                "joiners_at_risk": 0, "leavers_at_risk": 0,
            },
        }

    by_id = {p.id: p for p in subordinates}
    sub_ids = list(by_id.keys())
    today = _today()

    # ── Joiners: open onboarding processes ──
    onb_procs = (
        await db.execute(
            select(OnboardingProcess).where(
                OnboardingProcess.person_id.in_(sub_ids),
                OnboardingProcess.status.notin_(_TERMINAL_PROC),
            )
        )
    ).scalars().all()
    onb_blockers = await _pending_blocker_labels(
        db, OnboardingBlocker, OnboardingBlocker.process_id,
        [p.id for p in onb_procs],
    )

    joiners: list[dict[str, Any]] = []
    for proc in onb_procs:
        person = by_id.get(proc.person_id)
        if person is None:
            continue
        gaps = onb_blockers.get(proc.id, {"blocker": [], "advisory": []})
        hard_gap = len(gaps["blocker"])
        days_to_hire = (proc.hire_date - today).days if proc.hire_date else None
        at_risk = bool(
            proc.sla_breach_level in ("urgent", "breach")
            or (days_to_hire is not None and days_to_hire <= 0 and hard_gap > 0)
        )
        joiners.append({
            "process_id": str(proc.id),
            "person": _person_pack(person),
            "hire_date": proc.hire_date.isoformat() if proc.hire_date else None,
            "days_to_hire": days_to_hire,
            "status": proc.status,
            "sla_breach_level": proc.sla_breach_level,
            "hard_ready": proc.hard_ready_at is not None,
            "ready": proc.ready_at is not None,
            "pending_blocker_count": hard_gap,
            "pending_advisory_count": len(gaps["advisory"]),
            "pending_blocker_labels": gaps["blocker"][:12],
            "at_risk": at_risk,
        })

    # ── Leavers: open offboarding processes ──
    off_procs = (
        await db.execute(
            select(OffboardingProcess).where(
                OffboardingProcess.person_id.in_(sub_ids),
                OffboardingProcess.status.notin_(_TERMINAL_PROC),
            )
        )
    ).scalars().all()
    off_blockers = await _pending_blocker_labels(
        db, OffboardingBlocker, OffboardingBlocker.process_id,
        [p.id for p in off_procs],
    )

    leavers: list[dict[str, Any]] = []
    for proc in off_procs:
        person = by_id.get(proc.person_id)
        if person is None:
            continue
        gaps = off_blockers.get(proc.id, {"blocker": [], "advisory": []})
        hard_gap = len(gaps["blocker"])
        days_to_exit = (proc.leaver_date - today).days if proc.leaver_date else None
        # day-of-exit (<= 0) with a hard blocker still pending is the
        # highest-risk moment (employee walking out with live access) —
        # match the joiner side's <= 0 boundary, don't lean solely on SLA.
        at_risk = bool(
            proc.sla_breach_level in ("urgent", "breach")
            or (days_to_exit is not None and days_to_exit <= 0 and hard_gap > 0)
        )
        # Surface hardware-return status explicitly — the manager's
        # most common question on a leaver ("did they return the laptop?").
        hardware_labels = [
            lbl for lbl in gaps["blocker"]
            if "hardware" in lbl.lower() or "资产" in lbl or "设备" in lbl
        ]
        leavers.append({
            "process_id": str(proc.id),
            "person": _person_pack(person),
            "leaver_date": proc.leaver_date.isoformat() if proc.leaver_date else None,
            "days_to_exit": days_to_exit,
            "status": proc.status,
            "sla_breach_level": proc.sla_breach_level,
            "pending_blocker_count": hard_gap,
            "pending_advisory_count": len(gaps["advisory"]),
            "pending_blocker_labels": gaps["blocker"][:12],
            "hardware_pending": bool(hardware_labels),
            "default_handover_person_id": (
                str(proc.default_handover_person_id)
                if proc.default_handover_person_id else None
            ),
            "at_risk": at_risk,
        })

    joiners.sort(key=lambda r: (r["days_to_hire"] if r["days_to_hire"] is not None else 9999))
    leavers.sort(key=lambda r: (r["days_to_exit"] if r["days_to_exit"] is not None else 9999))

    return {
        "manager_id": str(manager.id),
        "subordinates_count": len(subordinates),
        "joiners": joiners,
        "leavers": leavers,
        "summary": {
            "joiners_count": len(joiners),
            "leavers_count": len(leavers),
            "joiners_at_risk": sum(1 for r in joiners if r["at_risk"]),
            "leavers_at_risk": sum(1 for r in leavers if r["at_risk"]),
        },
    }


def _person_pack(person: IdentityPerson) -> dict[str, Any]:
    return {
        "id": str(person.id),
        "display_name": person.display_name,
        "work_email": person.work_email,
        "employee_no": person.employee_no,
        "job_title": getattr(person, "job_title", None),
        "department": getattr(person, "department", None),
        "employment_status": person.employment_status,
    }


async def _pending_blocker_labels(
    db: AsyncSession,
    model,
    process_id_col,
    process_ids: list,
) -> dict:
    """Group pending blocker labels by process, split blocker vs advisory."""
    if not process_ids:
        return {}
    rows = (
        await db.execute(
            select(model).where(
                process_id_col.in_(process_ids),
                model.resolution == "pending",
            )
        )
    ).scalars().all()
    out: dict = {}
    for row in rows:
        bucket = out.setdefault(row.process_id, {"blocker": [], "advisory": []})
        key = "advisory" if getattr(row, "severity", "blocker") == "advisory" else "blocker"
        bucket[key].append(row.label)
    return out


async def build_team_access_attestation(
    db: AsyncSession,
    *,
    manager: IdentityPerson,
    include_left: bool = False,
) -> dict[str, Any]:
    """Manager-view access self-attestation status for the current open
    campaign across direct/transitive reports (who's done / pending /
    overdue). Read-only, no new push — the manager nudges informally. The
    attestation roll-up logic stays in the auditor domain; this wrapper
    only supplies the manager's subordinate scope."""
    from app.domains.auditor import access_certification_service as acs

    subs = await _list_subordinates(db, manager.id, include_left=include_left)
    data = await acs.build_team_attestation_status(
        db, person_ids=[p.id for p in subs],
    )
    data["manager_id"] = str(manager.id)
    data["subordinates_count"] = len(subs)
    return data


__all__ = [
    "build_team_access_attestation",
    "build_team_api_keys",
    "build_team_cloud",
    "build_team_footprint",
    "build_team_lifecycle",
]