"""Access certification engine — generalized periodic access review.

The IGA primitive (SailPoint/Saviynt) behind SOC2 CC6.3 / ISO A.5.15 &
A.5.18: a campaign snapshots a set of (person, entitlement) rows scoped
by ``campaign_type`` + ``scope_config``, assigns each a reviewer (the
holder's manager), and records keep/revoke decisions.

PR2 ships the ``software_seat``-by-department builder. The engine is a
registry of snapshot builders so cloud-grant / application / privileged-
role types plug in later without touching the campaign machinery. A
``revoke`` decision records the decision and flags the item for
remediation (``revocation_status='pending'``); fulfilment (calling the
owning domain's revoke path through a coordinator) is a separate step —
the attestation + decision IS the audit evidence.

Datetime discipline: naive UTC throughout (DateTime columns round-trip
naive on fresh load; mixing aware/naive subtraction 500s).
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import func, select, update
from sqlalchemy.exc import OperationalError
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.core.exceptions import AppException, NotFoundError
from app.core.org_scope import department_membership_subtree_clause
from app.core.txn import best_effort
from app.domains.asset.models import SoftwareLicensePlan, SoftwareSeatAllocation
from app.domains.auditor.constants import (
    PRIV_SCOPE_AWS,
    PRIV_SCOPE_FEISHU,
    PRIV_SCOPE_GITHUB,
    PRIV_SCOPE_GOOGLE,
    PRIV_SCOPE_HELIX,
)
from app.domains.auditor.integration_adapters import (
    AwsAuditorAdapter,
    FeishuAuditorAdapter,
    GitHubAuditorAdapter,
    GoogleAuditorAdapter,
    IntegrationMissingError,
)
from app.domains.auditor.models import (
    AccessCertification,
    AccessCertificationItem,
)
from app.identity.person_models import IdentityOrgMembership, IdentityPerson
from app.integrations.models import PlatformSetting

logger = logging.getLogger(__name__)

# User-creatable campaign types (drives the UI dropdown). privileged_role
# is created by the quarterly cron, not the UI, so it's intentionally
# absent here while still being a valid snapshot builder below.
CAMPAIGN_TYPES = ("software_seat", "identity_consumer")
_ESCALATION_GRACE_DAYS = 14
DECISION_PENDING = "pending"
DECISION_KEEP = "keep"
DECISION_REVOKE = "revoke"
# Terminal disposition for a self-attestation item whose holder never
# responded by the time the campaign is closed. Distinct from keep/revoke
# (no human decided) — lets an all-employee attestation campaign actually
# reach ``completed`` instead of blocking forever on non-responders.
DECISION_NO_RESPONSE = "no_response"
STATUS_IN_PROGRESS = "in_progress"
STATUS_COMPLETED = "completed"
STATUS_CANCELLED = "cancelled"

# ── Self-attestation (员工半年度访问自查) ─────────────────────────────
# A holder-attested campaign: instead of a central reviewer / the holder's
# manager deciding item-by-item, the access HOLDER reviews their OWN full
# access footprint in one batch ("还需要" = keep / "我不需要了" = revoke).
# Opened twice a year (June 1 / Dec 1) by a cron; the completed campaign is
# the SOC2 CC6.3 / ISO A.5.15 user-access-review evidence. Distinct from the
# manager/auditor-reviewed campaign types above — self_attestation items are
# decided ONLY through the batch self-service path, never ``decide_item``.
CAMPAIGN_TYPE_SELF_ATTESTATION = "self_attestation"

# ── "没人回答" 不再是终点 (访问对象闭环 PR5) ───────────────────────────
# A self-attestation closes by stamping every un-answered item
# ``no_response`` — access untouched, ``decided_by_person_id`` NULL. So
# the flow produced SOC2 evidence that a review happened while a person
# who never answered kept everything, and the same footprint came back
# six months later. This campaign type is the consequence: those items
# open a FOLLOW-UP whose reviewer is the holder's own manager, riding the
# existing ``decide_item`` + revocation pipeline rather than a second
# revoke path. **Nothing is auto-revoked** — the decision moves from
# somebody who did not answer to somebody who can.
CAMPAIGN_TYPE_NO_RESPONSE_FOLLOWUP = "no_response_followup"

#: ``report_only`` keeps the pre-PR5 behaviour (stamp + compliance
#: incident, nothing more). ``escalate_manager`` opens the follow-up.
NO_RESPONSE_POLICY_KEY = "access_review.self_attestation.no_response_policy"
NO_RESPONSE_REPORT_ONLY = "report_only"
NO_RESPONSE_ESCALATE_MANAGER = "escalate_manager"

#: Campaign types the SELF-SCOPED reviewer endpoint may decide.
#:
#: An allowlist, not a denylist, and deliberately narrow. Security review
#: 2026-09-05 found the first version of ``decide_assigned_review``
#: authorized purely on ``assigned_reviewer_person_id == caller`` with no
#: campaign-type restriction — which is a privilege escalation, because
#: ``_snapshot_identity_consumer`` sets BOTH ``person_id`` and
#: ``assigned_reviewer_person_id`` to the consumer's own owner. For that
#: campaign the "reviewer" IS the subject, so an ordinary employee could
#: have called this endpoint with ``keep`` on their own lapsed consumer
#: and had ``mark_identity_consumer_certified`` flip it back to ACTIVE
#: and clear ``recert_lapsed_at`` — undoing the very enforcement
#: ``lapse_disable_identity_consumers`` had applied, with no
#: security-role involvement. Those items were previously reachable only
#: through ``can_manage_security`` paths, and they stay that way.
SELF_SERVICE_REVIEWABLE_CAMPAIGN_TYPES = (CAMPAIGN_TYPE_NO_RESPONSE_FOLLOWUP,)

AT_SOFTWARE_SEAT = "software_seat"
AT_CLOUD_GRANT = "cloud_grant"
AT_USER_PERMISSION = "user_permission"
AT_SSO_GRANT = "sso_grant"
AT_API_KEY = "api_key"
# Personal BigQuery dataset grants (migration 606). A DAX ``google_user``
# binding is the person's own read access to a dataset — the most
# data-sensitive line on the footprint and, before 2026-09-05, the one line
# nobody was ever asked about.
AT_DATA_ACCESS_GRANT = "data_access_grant"

# The six access types enumerated into a person's footprint snapshot.
SELF_ATTESTATION_ACCESS_TYPES = (
    AT_SOFTWARE_SEAT, AT_CLOUD_GRANT, AT_USER_PERMISSION,
    AT_SSO_GRANT, AT_API_KEY, AT_DATA_ACCESS_GRANT,
)
# Types whose "我不需要了" surfaces a self-service removal action. API keys
# are attest-only (the org deliberately does not auto-reclaim key risk —
# they are reminded to confirm, never removed by this flow). SSO grants
# surface the action but route to an IT handover disposition (no clean
# holder self-revoke exists — only handover-to-successor / admin revoke).
# DAX grants self-remove through the holder's own release path.
SELF_REMOVABLE_ACCESS_TYPES = (
    AT_SOFTWARE_SEAT, AT_CLOUD_GRANT, AT_USER_PERMISSION, AT_SSO_GRANT,
    AT_DATA_ACCESS_GRANT,
)
# Higher-sensitivity types sort first on the employee review surface.
_ELEVATED_ACCESS_TYPES = (AT_CLOUD_GRANT, AT_API_KEY, AT_SSO_GRANT, AT_DATA_ACCESS_GRANT)
# Three notifications, three keys. A ``template_key`` resolves to exactly ONE
# piece of copy, and these say different things: "please confirm your access"
# (campaign opened), "you are past the deadline" (holder nudge) and "your
# report is past the deadline" (manager heads-up). They all used to share the
# key below — harmless only while it was unregistered and every recipient got
# the call site's hardcoded Chinese; registering one template for all three
# would have retitled the campaign-open card "已逾期". Same split, same reason,
# as ``procurement.reminder`` → ``procurement.reminder.*``.
#
# The open key keeps the original spelling because the re-run guard in
# ``_notify_self_attestation_holders`` looks back on
# ``(template_key, related_resource_id)`` rows already written in prod.
SELF_ATTESTATION_TEMPLATE_KEY = "access_review.self_attestation"
SELF_ATTESTATION_OVERDUE_TEMPLATE_KEY = "access_review.self_attestation.overdue"
SELF_ATTESTATION_OVERDUE_MANAGER_TEMPLATE_KEY = (
    "access_review.self_attestation.overdue_manager"
)


def _naive_now() -> datetime:
    return datetime.now(UTC).replace(tzinfo=None)


def _to_naive(dt: datetime | None) -> datetime | None:
    if dt is None:
        return None
    if dt.tzinfo is not None:
        return dt.astimezone(UTC).replace(tzinfo=None)
    return dt


# ════════════════════════════════════════════════════════════════════
# Snapshot builders (campaign_type → rows)
# ════════════════════════════════════════════════════════════════════


async def _snapshot_software_seat(
    db: AsyncSession,
    *,
    certification: AccessCertification,
    snapshot_at: datetime,
) -> int:
    """One item per active software seat held by a person in the scoped
    department subtree. Reviewer = the holder's manager.
    """
    scope = certification.scope_config or {}
    department = (scope.get("department") or "").strip()
    if not department:
        raise AppException(
            code="CERT_SCOPE_REQUIRED",
            message="software_seat campaigns require scope_config.department",
            status_code=422,
        )

    q = (
        select(SoftwareSeatAllocation, SoftwareLicensePlan, IdentityPerson)
        .join(
            SoftwareLicensePlan,
            SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id,
        )
        .join(
            IdentityPerson,
            IdentityPerson.id == SoftwareSeatAllocation.user_id,
        )
        .join(
            IdentityOrgMembership,
            IdentityOrgMembership.person_id == IdentityPerson.id,
        )
        .where(
            SoftwareSeatAllocation.status == "active",
            IdentityOrgMembership.is_primary.is_(True),
            IdentityOrgMembership.deleted_at.is_(None),
            IdentityPerson.deleted_at.is_(None),
            department_membership_subtree_clause(department),
        )
    )
    rows = (await db.execute(q)).all()
    n = 0
    for seat, plan, person in rows:
        db.add(AccessCertificationItem(
            id=uuid.uuid4(),
            certification_id=certification.id,
            person_id=person.id,
            holder_email=person.work_email,
            holder_display_name=person.display_name,
            access_type="software_seat",
            access_ref=str(seat.id),
            access_label=plan.name,
            resource_name=plan.name,
            assigned_reviewer_person_id=person.manager_person_id,
            decided_at_deadline=certification.due_at,
            decision=DECISION_PENDING,
            snapshot_at=snapshot_at,
            raw_payload={
                "seat_id": str(seat.id),
                "plan_id": str(seat.plan_id),
                "plan_name": plan.name,
                "person_id": str(person.id),
                "work_email": person.work_email,
            },
        ))
        n += 1
    await db.flush()
    return n


async def _snapshot_privileged_role(
    db: AsyncSession,
    *,
    certification: AccessCertification,
    snapshot_at: datetime,
) -> int:
    """One item per privileged-role holder across every connector:
    Helix admins, Google admins, AWS root accounts, GitHub org owners,
    Feishu admins, Atlassian org admins. Holders are mostly external
    (no Helix person), so reviewers are assigned by the auditor — items
    start with no reviewer but carry the campaign deadline so the
    escalation sweep still fires.
    """
    n = 0
    deadline = certification.due_at

    def _add(*, scope, role_name, external_id, email, display_name,
             person_id, raw):
        db.add(AccessCertificationItem(
            id=uuid.uuid4(),
            certification_id=certification.id,
            person_id=person_id,
            holder_email=email,
            holder_display_name=display_name,
            access_type="privileged_role",
            access_ref=external_id,
            access_label=f"{scope}/{role_name}",
            resource_name=scope,
            assigned_reviewer_person_id=None,
            decided_at_deadline=deadline,
            decision=DECISION_PENDING,
            snapshot_at=snapshot_at,
            raw_payload=raw,
        ))

    # ── Helix privileged role holders ────────────────────────────────
    from app.auth.models import AuthRole, AuthUserRole

    priv_role_names = (
        "it_admin", "global_admin", "asset_admin", "asset_global_admin",
        "cloud_admin", "security_auditor",
    )
    # Tombstoned holders are excluded: a soft-deleted person keeps
    # employment_status='active', so without this filter a purged
    # duplicate showed up as a sitting global_admin, and the reviewer's
    # only sane answer (revoke) landed in the manual_required dead
    # queue against a person who no longer exists. They are not merely
    # dropped — the leftover auth_user_roles row is a real hygiene gap,
    # so it is logged for the RBAC cleanup cron to pick up.
    ghost_rows = (await db.execute(
        select(func.count())
        .select_from(AuthUserRole)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .join(IdentityPerson, IdentityPerson.id == AuthUserRole.user_id)
        .where(
            AuthRole.name.in_(priv_role_names),
            IdentityPerson.deleted_at.is_not(None),
        )
    )).scalar_one()
    if ghost_rows:
        logger.warning(
            "privileged snapshot skipped %d role row(s) held by "
            "soft-deleted persons — auth_user_roles needs cleanup",
            int(ghost_rows),
        )

    helix_rows = (await db.execute(
        select(AuthUserRole, AuthRole, IdentityPerson)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .join(IdentityPerson, IdentityPerson.id == AuthUserRole.user_id)
        .where(
            AuthRole.name.in_(priv_role_names),
            IdentityPerson.deleted_at.is_(None),
        )
    )).all()
    for _ur, role_row, person_row in helix_rows:
        _add(scope=PRIV_SCOPE_HELIX, role_name=role_row.name,
             external_id=str(person_row.id), email=person_row.work_email,
             display_name=person_row.display_name, person_id=person_row.id,
             raw={"person_id": str(person_row.id), "role": role_row.name})
        n += 1

    # ── External connectors (degrade gracefully if not configured) ───
    try:
        for a in await GoogleAuditorAdapter.list_admin_role_assignments(db):
            if not a.role_name:
                continue
            _add(scope=PRIV_SCOPE_GOOGLE, role_name=a.role_name,
                 external_id=a.assigned_to_user_id, email=a.assigned_to_email,
                 display_name=None, person_id=None, raw=a.raw)
            n += 1
    except IntegrationMissingError as exc:
        logger.warning("Google admin snapshot skipped: %s", exc)

    try:
        for r in await AwsAuditorAdapter.list_root_account_holders(db):
            _add(scope=PRIV_SCOPE_AWS, role_name="root_account",
                 external_id=r.account_id, email=r.root_email,
                 display_name=r.account_alias, person_id=None, raw=r.raw)
            n += 1
    except IntegrationMissingError as exc:
        logger.warning("AWS root snapshot skipped: %s", exc)

    try:
        for a in await GitHubAuditorAdapter.list_organization_admins(db):
            _add(scope=PRIV_SCOPE_GITHUB, role_name=a.role,
                 external_id=a.user_login, email=a.user_email,
                 display_name=a.user_login, person_id=None, raw=a.raw)
            n += 1
    except IntegrationMissingError as exc:
        logger.warning("GitHub admin snapshot skipped: %s", exc)

    try:
        for a in await FeishuAuditorAdapter.list_admin_role_holders(db):
            _add(scope=PRIV_SCOPE_FEISHU, role_name=a.role_name,
                 external_id=a.user_id, email=a.user_email,
                 display_name=None, person_id=None, raw=a.raw)
            n += 1
    except IntegrationMissingError as exc:
        logger.warning("Feishu admin snapshot skipped: %s", exc)

    try:
        from app.domains.auditor.constants import PRIV_SCOPE_ATLASSIAN
        from app.domains.auditor.integration_adapters import (
            AtlassianAuditorAdapter,
        )
        for a in await AtlassianAuditorAdapter.list_organization_admins(db):
            _add(scope=PRIV_SCOPE_ATLASSIAN, role_name="org-admin",
                 external_id=a.user_account_id, email=a.user_email,
                 display_name=a.user_name, person_id=None, raw=a.raw)
            n += 1
    except IntegrationMissingError as exc:
        logger.warning("Atlassian admin snapshot skipped: %s", exc)

    await db.flush()
    return n


async def _snapshot_identity_consumer(
    db: AsyncSession,
    *,
    certification: AccessCertification,
    snapshot_at: datetime,
) -> int:
    """One item per active identity consumer (the outbound data-pull
    surface, PLATFORM §10.6). Reviewer hint = the consumer owner; the
    central auditor decides. ``raw_payload`` carries a usage signal
    (requests in the last 90 days + dormancy flag) so the reviewer can
    spot consumers to decommission. A ``revoke`` decision disables the
    consumer (kills PII egress) via the coordinator; ``keep`` stamps
    ``last_certified_at``."""
    from datetime import timedelta as _td

    from app.identity.consumers.models import (
        CONSUMER_STATUS_ACTIVE,
        IdentityConsumer,
        IdentityConsumerUsageDaily,
    )

    consumers = (await db.execute(
        select(IdentityConsumer).where(
            IdentityConsumer.status == CONSUMER_STATUS_ACTIVE,
            IdentityConsumer.deleted_at.is_(None),
        )
    )).scalars().all()
    if not consumers:
        return 0

    ids = [c.id for c in consumers]
    since = snapshot_at.date() - _td(days=90)
    usage_rows = (await db.execute(
        select(
            IdentityConsumerUsageDaily.consumer_id,
            func.coalesce(func.sum(IdentityConsumerUsageDaily.request_count), 0),
        )
        .where(
            IdentityConsumerUsageDaily.consumer_id.in_(ids),
            IdentityConsumerUsageDaily.usage_date >= since,
        )
        .group_by(IdentityConsumerUsageDaily.consumer_id)
    )).all()
    usage_by_consumer = {cid: int(n) for cid, n in usage_rows}

    owner_ids = {c.owner_person_id for c in consumers}
    owners: dict[uuid.UUID, IdentityPerson] = {}
    if owner_ids:
        rows = (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids))
        )).scalars().all()
        owners = {p.id: p for p in rows}

    n = 0
    for c in consumers:
        owner = owners.get(c.owner_person_id)
        reqs = usage_by_consumer.get(c.id, 0)
        db.add(AccessCertificationItem(
            id=uuid.uuid4(),
            certification_id=certification.id,
            person_id=c.owner_person_id,
            holder_email=owner.work_email if owner else None,
            holder_display_name=owner.display_name if owner else None,
            access_type="identity_consumer",
            access_ref=str(c.id),
            access_label=c.name,
            resource_name=c.slug,
            assigned_reviewer_person_id=c.owner_person_id,
            decided_at_deadline=certification.due_at,
            decision=DECISION_PENDING,
            snapshot_at=snapshot_at,
            raw_payload={
                "consumer_id": str(c.id),
                "slug": c.slug,
                "scopes": list(c.scopes or []),
                "requests_90d": reqs,
                "dormant": reqs == 0,
                "last_certified_at": (
                    _to_naive(c.last_certified_at).isoformat()
                    if c.last_certified_at else None
                ),
            },
        ))
        n += 1
    await db.flush()
    return n


def _sensitivity_for(access_type: str) -> str:
    return "elevated" if access_type in _ELEVATED_ACCESS_TYPES else "standard"


async def _snapshot_self_attestation(
    db: AsyncSession,
    *,
    certification: AccessCertification,
    snapshot_at: datetime,
) -> int:
    """Enumerate every active employee's FULL access footprint — software
    seats, cloud access grants, software permissions, SSO grants, API keys
    — one item per (person, entitlement). Reviewer = the holder themselves
    (``assigned_reviewer_person_id = person_id``), so the holder attests
    their own access in a single batch.

    Heterogeneous holder-id column types (some ``Uuid``, some ``String(36)``)
    are reconciled by loading active persons into a ``str(id)``-keyed map and
    mapping each entitlement row's holder via ``str(...)`` — never a
    cross-type SQL JOIN (asyncmy silently returns 0 rows on a ``Uuid`` ↔
    ``String`` join / bind). Persons with zero entitlements get no items and
    are therefore never notified.
    """
    from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider
    from app.domains.asset.models import AssetSoftwareCatalog
    from app.domains.cmdb.models import (
        GRANT_STATUS_ACTIVE,
        CmdbCloudAccessGrant,
        CmdbCloudScope,
    )
    from app.domains.permission.models import (
        PERM_ACTIVE,
        SoftwarePermission,
        UserPermission,
    )
    from app.domains.sso.models import SSO_GRANT_STATUS_APPROVED, SsoGrant

    persons = {
        str(p.id): p
        for p in (await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.employment_status == "active",
                IdentityPerson.deleted_at.is_(None),
            )
        )).scalars().all()
    }
    if not persons:
        return 0

    deadline = certification.due_at
    n = 0

    def _add(*, person, access_type, access_ref, access_label,
             resource_name, granted_at, last_used_at, raw):
        full_raw = {
            **raw,
            "access_type": access_type,
            "sensitivity": _sensitivity_for(access_type),
            "removable": access_type in SELF_REMOVABLE_ACCESS_TYPES,
            "granted_at": (
                _to_naive(granted_at).isoformat() if granted_at else None
            ),
            "last_used_at": (
                _to_naive(last_used_at).isoformat() if last_used_at else None
            ),
        }
        db.add(AccessCertificationItem(
            id=uuid.uuid4(),
            certification_id=certification.id,
            person_id=person.id,
            holder_email=person.work_email,
            holder_display_name=person.display_name,
            access_type=access_type,
            access_ref=access_ref,
            access_label=access_label[:255],
            resource_name=(resource_name or None) and resource_name[:255],
            assigned_reviewer_person_id=person.id,
            decided_at_deadline=deadline,
            decision=DECISION_PENDING,
            snapshot_at=snapshot_at,
            raw_payload=full_raw,
        ))

    # ── 1. Software seats ────────────────────────────────────────────
    seat_rows = (await db.execute(
        select(SoftwareSeatAllocation, SoftwareLicensePlan, AssetSoftwareCatalog)
        .join(SoftwareLicensePlan,
              SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id)
        .join(AssetSoftwareCatalog,
              AssetSoftwareCatalog.id == SoftwareLicensePlan.software_id)
        .where(SoftwareSeatAllocation.status == "active")
    )).all()
    for seat, plan, sw in seat_rows:
        person = persons.get(str(seat.user_id))
        if person is None:
            continue
        _add(
            person=person, access_type=AT_SOFTWARE_SEAT,
            access_ref=str(seat.id),
            access_label=f"{sw.name} · {plan.name}",
            resource_name=sw.name, granted_at=seat.allocated_at,
            last_used_at=None,
            raw={"seat_id": str(seat.id), "plan_id": str(seat.plan_id),
                 "plan_name": plan.name, "software": sw.name},
        )
        n += 1

    # ── 2. Cloud access grants ───────────────────────────────────────
    scope_names = {
        str(sid): dn
        for sid, dn in (await db.execute(
            select(CmdbCloudScope.id, CmdbCloudScope.display_name)
        )).all()
    }
    grant_rows = (await db.execute(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE,
            CmdbCloudAccessGrant.person_id.is_not(None),
        )
    )).scalars().all()
    for g in grant_rows:
        person = persons.get(str(g.person_id))
        if person is None:
            continue
        scope_label = scope_names.get(str(g.scope_id)) or g.scope_id
        role_label = g.role_display_name or g.role_key
        _add(
            person=person, access_type=AT_CLOUD_GRANT, access_ref=str(g.id),
            access_label=f"{scope_label} · {role_label}",
            resource_name=scope_label, granted_at=g.granted_at,
            last_used_at=g.last_login_at,
            raw={"grant_id": str(g.id), "scope_id": str(g.scope_id),
                 "role_key": g.role_key, "expires_at": (
                     _to_naive(g.expires_at).isoformat()
                     if g.expires_at else None)},
        )
        n += 1

    # ── 3. Software permissions ──────────────────────────────────────
    perm_rows = (await db.execute(
        select(UserPermission, SoftwarePermission, AssetSoftwareCatalog)
        .join(SoftwarePermission,
              SoftwarePermission.id == UserPermission.permission_id)
        .join(AssetSoftwareCatalog,
              AssetSoftwareCatalog.id == SoftwarePermission.software_id)
        .where(UserPermission.status == PERM_ACTIVE)
    )).all()
    for up, perm, sw in perm_rows:
        person = persons.get(str(up.user_id))
        if person is None:
            continue
        _add(
            person=person, access_type=AT_USER_PERMISSION,
            access_ref=str(up.id),
            access_label=f"{sw.name} · {perm.display_name}",
            resource_name=sw.name, granted_at=up.granted_at,
            last_used_at=None,
            raw={"user_permission_id": str(up.id),
                 "permission_key": perm.permission_key, "software": sw.name},
        )
        n += 1

    # ── 4. SSO grants ────────────────────────────────────────────────
    sso_rows = (await db.execute(
        select(SsoGrant).where(
            SsoGrant.status == SSO_GRANT_STATUS_APPROVED,
        )
    )).scalars().all()
    for s in sso_rows:
        person = persons.get(str(s.person_id))
        if person is None:
            continue
        _add(
            person=person, access_type=AT_SSO_GRANT, access_ref=str(s.id),
            access_label=f"{s.sso_provider} · {s.system_name}",
            resource_name=s.system_name, granted_at=s.granted_at,
            last_used_at=None,
            raw={"sso_grant_id": str(s.id), "sso_provider": s.sso_provider,
                 "system_name": s.system_name},
        )
        n += 1

    # ── 5. API keys (attest-only) ────────────────────────────────────
    provider_names = {
        str(pid): name
        for pid, name in (await db.execute(
            select(ApiKeyProvider.id, ApiKeyProvider.name)
        )).all()
    }
    ak_rows = (await db.execute(
        select(ApiKeyAssignment).where(
            ApiKeyAssignment.status == "active",
            ApiKeyAssignment.assigned_to.is_not(None),
        )
    )).scalars().all()
    for ak in ak_rows:
        person = persons.get(str(ak.assigned_to))
        if person is None:
            continue
        prov = provider_names.get(str(ak.provider_id)) or "API Key"
        _add(
            person=person, access_type=AT_API_KEY, access_ref=str(ak.id),
            access_label=f"{prov} · {ak.project_name}",
            resource_name=prov, granted_at=ak.created_at,
            last_used_at=None,
            raw={"assignment_id": str(ak.id), "provider": prov,
                 "request_number": ak.request_number},
        )
        n += 1

    # ── 6. DAX personal dataset grants (migration 606) ───────────────
    from app.domains.data_access import constants as dax_c
    from app.domains.data_access.models import (
        BINDING_STATUS_ACTIVE as DAX_BINDING_ACTIVE,
    )
    from app.domains.data_access.models import (
        DataAccessBinding,
        DataAccessResource,
    )

    dax_rows = (await db.execute(
        select(DataAccessBinding, DataAccessResource)
        .join(DataAccessResource, DataAccessResource.id == DataAccessBinding.resource_id)
        .where(
            DataAccessBinding.status == DAX_BINDING_ACTIVE,
            DataAccessBinding.principal_type_snapshot == dax_c.PRINCIPAL_TYPE_GOOGLE_USER,
        )
    )).all()
    for b, res in dax_rows:
        person = persons.get(str(b.principal_subject_snapshot))
        if person is None:
            continue
        locator = ".".join(filter(None, (res.project_id, res.dataset_id, res.table_id)))
        expires_naive = _to_naive(b.expires_at) if b.expires_at else None
        _add(
            person=person, access_type=AT_DATA_ACCESS_GRANT, access_ref=str(b.id),
            access_label=f"BigQuery · {locator} · {b.entitlement_snapshot}",
            resource_name=locator, granted_at=b.granted_at,
            last_used_at=None,
            raw={"binding_id": str(b.id), "resource_id": str(res.id),
                 "locator": locator, "entitlement": b.entitlement_snapshot,
                 "expires_at": expires_naive.isoformat() if expires_naive else None},
        )
        n += 1

    await db.flush()
    return n


_SNAPSHOT_BUILDERS = {
    "software_seat": _snapshot_software_seat,
    "privileged_role": _snapshot_privileged_role,
    "identity_consumer": _snapshot_identity_consumer,
    CAMPAIGN_TYPE_SELF_ATTESTATION: _snapshot_self_attestation,
}


# ════════════════════════════════════════════════════════════════════
# Campaign lifecycle
# ════════════════════════════════════════════════════════════════════


async def create_campaign(
    db: AsyncSession,
    *,
    label: str,
    campaign_type: str,
    scope_config: dict[str, Any] | None,
    due_at: datetime,
    actor_person_id: uuid.UUID | None,
    trace_id: uuid.UUID | None = None,
    _allow_self_attestation: bool = False,
) -> AccessCertification:
    if campaign_type not in _SNAPSHOT_BUILDERS:
        raise AppException(
            code="UNSUPPORTED_CAMPAIGN_TYPE",
            message=(
                f"campaign_type must be one of {sorted(_SNAPSHOT_BUILDERS)}"
            ),
            status_code=422,
        )
    if (
        campaign_type == CAMPAIGN_TYPE_SELF_ATTESTATION
        and not _allow_self_attestation
    ):
        # This generic entry point has none of the three things a
        # self-attestation campaign needs, and every one of them is
        # load-bearing (#2164): the Redis lock + label idempotency that
        # stop a second campaign co-existing, and the holder fan-out.
        # Opened from here, thousands of items get snapshotted with NO
        # notification, ``my_open_self_attestation`` (newest in_progress
        # wins) hides the older campaign from employees, and 14 days
        # past the deadline ``escalate_overdue`` opens an incident per
        # employee and chases everyone about a self-check they were
        # never told had started.
        raise AppException(
            code="USE_SELF_ATTESTATION_OPENER",
            message=(
                "self_attestation campaigns must be opened via "
                "POST /compliance/certifications/self-attestation/open-now "
                "(or the半年 cron) — that path holds the open lock, is "
                "idempotent by period label, and notifies every holder"
            ),
            status_code=422,
        )
    started_at = _naive_now()
    cert = AccessCertification(
        id=uuid.uuid4(),
        label=label,
        campaign_type=campaign_type,
        scope_config=scope_config or {},
        status=STATUS_IN_PROGRESS,
        started_at=started_at,
        due_at=_to_naive(due_at),
        created_by_person_id=actor_person_id,
    )
    db.add(cert)
    await db.flush()

    builder = _SNAPSHOT_BUILDERS[campaign_type]
    count = await builder(db, certification=cert, snapshot_at=started_at)
    cert.snapshot_item_count = count

    await AuditService(db).log(
        trace_id=trace_id or uuid.uuid4(),
        actor_type="user" if actor_person_id else "system",
        actor_user_id=actor_person_id,
        action="auditor.access_certification.create",
        resource_type="access_certifications",
        resource_id=str(cert.id),
        status="success",
        detail={
            "label": label,
            "campaign_type": campaign_type,
            "scope_config": scope_config or {},
            "snapshot_item_count": count,
        },
    )
    return cert


async def decide_item(
    db: AsyncSession,
    *,
    item_id: uuid.UUID,
    decision: str,
    note: str | None,
    actor_person_id: uuid.UUID,
    trace_id: uuid.UUID | None = None,
) -> AccessCertificationItem:
    if decision not in (DECISION_KEEP, DECISION_REVOKE):
        raise AppException(
            code="INVALID_DECISION",
            message="decision must be 'keep' or 'revoke'",
            status_code=422,
        )
    # FOR UPDATE: the pending guard below is a read-then-write, and the
    # counter bump after it is unconditional. Two reviewers (or HTTP +
    # MCP) deciding the same item at once both saw ``pending`` in their
    # own snapshot, both passed the guard, and each added 1 — pushing
    # ``decided_count`` past ``snapshot_item_count`` with no reconcile
    # cron to heal it (#2162).
    #
    # ``populate_existing=True`` is NOT optional here, and it is the half
    # that is easy to leave off because the lock still looks taken.
    # ``Session.get(..., with_for_update=...)`` does skip the identity-map
    # shortcut, so the ``SELECT ... FOR UPDATE`` really is emitted and the
    # row really is locked — but WITHOUT ``populate_existing`` the fetched
    # values are DISCARDED for an instance already in the session, and the
    # guard below then reads the stale in-memory ``decision``. That turns
    # this lock back into exactly the #2162 double-decide it exists to
    # stop, for any caller that touched the item first — which is what
    # ``decide_assigned_review`` does, because its authorization check has
    # to read ``assigned_reviewer_person_id`` before delegating here.
    item = await db.get(
        AccessCertificationItem, item_id,
        with_for_update=True, populate_existing=True,
    )
    if item is None:
        raise NotFoundError("AccessCertificationItem", str(item_id))
    cert = await db.get(AccessCertification, item.certification_id)
    if cert is None:
        raise NotFoundError("AccessCertification", str(item.certification_id))
    if cert.campaign_type == CAMPAIGN_TYPE_SELF_ATTESTATION:
        # Self-attestation items are decided by the holder via the batch
        # self-service path (``submit_self_attestation``), never the
        # manager/auditor item-by-item endpoint.
        raise AppException(
            code="SELF_ATTESTATION_ITEM",
            message=(
                "self_attestation items are decided by the holder via "
                "/me/access-review, not the reviewer decide endpoint"
            ),
            status_code=409,
        )
    if cert.status != STATUS_IN_PROGRESS:
        raise AppException(
            code="CAMPAIGN_NOT_OPEN",
            message=f"campaign {cert.id} is {cert.status}",
            status_code=409,
        )
    if item.decision != DECISION_PENDING:
        raise AppException(
            code="DECISION_ALREADY_RECORDED",
            message=f"item {item_id} already decided ({item.decision})",
            status_code=409,
        )

    now = _naive_now()
    item.decision = decision
    item.decided_by_person_id = actor_person_id
    item.decided_at = now
    item.decision_note = note
    # Atomic SQL increment (never read-modify-write in Python) so concurrent
    # reviewers on the same campaign can't lose-update the shared counters.
    await db.execute(
        update(AccessCertification)
        .where(AccessCertification.id == cert.id)
        .values(
            decided_count=AccessCertification.decided_count + 1,
            revoked_count=(
                AccessCertification.revoked_count
                + (1 if decision == DECISION_REVOKE else 0)
            ),
        )
    )
    if decision == DECISION_REVOKE:
        # The attestation is the audit evidence AND the directive to
        # remove access. Trigger real fulfilment via the coordinator
        # (auditor must not reach into asset/provisioning directly). If
        # the downstream enqueue hiccups we still record the decision and
        # leave the item ``pending`` — the revoke-fulfil cron retries —
        # so a vendor blip never loses the reviewer's attestation.
        from app.core.coordinators.access_certification_coordinator import (
            REVOCATION_PENDING,
            enqueue_revoke_for_cert_item,
        )
        # SAVEPOINT-isolate (2026-08-06): the promise above — "a vendor blip
        # never loses the reviewer's attestation" — needs the session to
        # survive. Without it the failed enqueue deactivated the transaction
        # and the AuditService.log below (the attestation evidence itself)
        # raised, 500-ing the decide endpoint and losing the decision.
        async with best_effort(
            db, label=f"access-cert revoke enqueue for item {item.id}", log=logger,
        ) as enqueue_step:
            await enqueue_revoke_for_cert_item(
                db, item=item, actor_user_id=actor_person_id,
                trace_id=str(trace_id or uuid.uuid4()),
            )
        if not enqueue_step.ok:
            logger.error(
                "access-cert revoke enqueue failed for item %s (%s); "
                "left pending for the fulfil cron", item.id,
                enqueue_step.reason(),
            )
            item.revocation_status = REVOCATION_PENDING
    elif decision == DECISION_KEEP and item.access_type == "identity_consumer":
        # Keep = re-attested. Stamp the consumer's last_certified_at and,
        # if it was auto-disabled by a recert lapse, re-enable it. Routed
        # through the coordinator (auditor must not write the consumer
        # row directly). Best-effort: a hiccup must not lose the keep
        # attestation (the audit row below is the evidence).
        from app.core.coordinators.access_certification_coordinator import (
            mark_identity_consumer_certified,
        )
        # SAVEPOINT-isolate — same rationale: the keep attestation's evidence
        # is the audit row below, and it must still be writable.
        async with best_effort(
            db,
            label=f"identity-consumer recert keep for item {item.id}",
            log=logger,
        ) as keep_step:
            await mark_identity_consumer_certified(
                db, item=item, trace_id=str(trace_id or uuid.uuid4()),
            )
        if not keep_step.ok:
            logger.error(
                "identity-consumer recert keep side-effect failed for item "
                "%s: %s", item.id, keep_step.reason(),
            )

    await AuditService(db).log(
        trace_id=trace_id or uuid.uuid4(),
        actor_type="user",
        actor_user_id=actor_person_id,
        action=f"auditor.access_certification.decide.{decision}",
        resource_type="access_certification_items",
        resource_id=str(item.id),
        status="success",
        detail={
            "certification_id": str(cert.id),
            "access_type": item.access_type,
            "access_ref": item.access_ref,
            "decision": decision,
        },
    )
    return item



async def _no_response_policy(db: AsyncSession) -> str:
    row = await db.get(PlatformSetting, NO_RESPONSE_POLICY_KEY)
    value = row.value if row is not None else None
    text = str(value or "").strip().lower()
    return text if text in (
        NO_RESPONSE_REPORT_ONLY, NO_RESPONSE_ESCALATE_MANAGER,
    ) else NO_RESPONSE_REPORT_ONLY




def _followup_reviewer_for(holder: IdentityPerson | None) -> uuid.UUID | None:
    """The manager who should rule on this holder's non-response, if any.

    ``None`` when there is nobody appropriate — the item then stays
    unassigned and routes to the auditor console, which is the same
    fallback the manager-reviewed campaign types use.

    A holder whose ``manager_person_id`` is themselves (self-managed
    records exist for executives and for HRIS edge cases) is NOT given
    their own non-response to rule on: that would turn "you did not
    answer, so your manager decides" into "you did not answer, so you
    decide", which is the review having no teeth at all. The decide
    endpoint refuses this case too — belt and braces, because the row
    simply never existing is the stronger guarantee.
    """
    if holder is None or holder.manager_person_id is None:
        return None
    if holder.manager_person_id == holder.id:
        return None
    return holder.manager_person_id

def _followup_window(cert: AccessCertification) -> timedelta:
    """The follow-up gets the same window the source campaign had.

    Falls back to 14 days when either end is missing. A follow-up with no
    deadline is a list nobody is ever late for, which is exactly how the
    original problem looked.
    """
    started = _to_naive(cert.started_at)
    due = _to_naive(cert.due_at)
    if started is None or due is None:
        return timedelta(days=14)
    span = due - started
    return span if span > timedelta(0) else timedelta(days=14)

async def _open_no_response_followup(
    db: AsyncSession,
    *,
    cert: AccessCertification,
    actor_person_id: uuid.UUID | None,
    trace_id: uuid.UUID | None,
) -> AccessCertification | None:
    """Open a manager-reviewed follow-up over this campaign's non-responders.

    Returns the new campaign, or ``None`` when the policy is
    ``report_only`` (the default) or nobody failed to answer.

    Why a real campaign rather than a bespoke worklist: ``decide_item``,
    the revocation pipeline, overdue escalation, the auditor console and
    the compliance-incident path all already exist and all key off a
    campaign. A second mechanism would need its own version of each, and
    the two would drift — the reason this repo has one provisioning
    queue rather than one per caller.

    The reviewer is the holder's manager. When a holder has no manager on
    record the item is left unassigned, which routes it to the auditor
    console — the same fallback the manager-reviewed campaign types use.
    A leaver's items are skipped entirely: their access is offboarding's
    business, not a review's.
    """
    policy = await _no_response_policy(db)
    if policy != NO_RESPONSE_ESCALATE_MANAGER:
        return None

    rows = (await db.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert.id,
            AccessCertificationItem.decision == DECISION_NO_RESPONSE,
        )
    )).scalars().all()
    if not rows:
        return None

    person_ids = sorted({r.person_id for r in rows if r.person_id})
    people = {
        p.id: p
        for p in (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(person_ids))
        )).scalars().all()
    } if person_ids else {}

    followup = AccessCertification(
        id=uuid.uuid4(),
        label=f"{cert.label} · 未回答补审",
        campaign_type=CAMPAIGN_TYPE_NO_RESPONSE_FOLLOWUP,
        scope_config={"source_certification_id": str(cert.id)},
        status=STATUS_IN_PROGRESS,
        started_at=_naive_now(),
        # Same window the source campaign got. A follow-up with no
        # deadline is a list nobody is ever late for, which is how the
        # original problem looked.
        due_at=_naive_now() + _followup_window(cert),
        created_by_person_id=actor_person_id,
        notes=(
            f"自动开出:{cert.label} 收尾时有 {len(rows)} 项没人回答。"
            "决定权交给持有人的直属经理;没有任何权限被自动撤销。"
        ),
    )
    db.add(followup)
    await db.flush()

    copied = 0
    skipped_leaver = 0
    for row in rows:
        holder = people.get(row.person_id) if row.person_id else None
        if holder is not None and holder.employment_status != "active":
            # Not a review question any more — offboarding owns a
            # leaver's access, and asking a manager to re-attest it would
            # duplicate (and race) that flow.
            skipped_leaver += 1
            continue
        db.add(AccessCertificationItem(
            id=uuid.uuid4(),
            certification_id=followup.id,
            person_id=row.person_id,
            holder_email=row.holder_email,
            holder_display_name=row.holder_display_name,
            access_type=row.access_type,
            access_ref=row.access_ref,
            access_label=row.access_label,
            resource_name=row.resource_name,
            assigned_reviewer_person_id=_followup_reviewer_for(holder),
            decision=DECISION_PENDING,
            snapshot_at=_naive_now(),
            raw_payload={
                **(row.raw_payload or {}),
                "no_response_source_item_id": str(row.id),
                "no_response_source_campaign": str(cert.id),
            },
        ))
        copied += 1

    followup.snapshot_item_count = copied
    await db.flush()

    if copied == 0:
        # Everyone who failed to answer has since left. Nothing to review
        # — drop the empty campaign rather than leaving a zero-item shell
        # that reads like an open obligation.
        await db.delete(followup)
        await db.flush()
        return None

    await AuditService(db).log(
        trace_id=trace_id or uuid.uuid4(),
        actor_type="system",
        actor_user_id=actor_person_id,
        action="auditor.access_certification.no_response_followup_opened",
        resource_type="access_certifications",
        resource_id=str(followup.id),
        status="success",
        detail={
            "source_certification_id": str(cert.id),
            "no_response_items": len(rows),
            "followup_items": copied,
            "skipped_leaver_items": skipped_leaver,
        },
    )
    logger.info(
        "no_response_followup: cert=%s -> %s items=%s (skipped_leaver=%s)",
        cert.id, followup.id, copied, skipped_leaver,
    )
    # Telling the reviewer. Setting ``assigned_reviewer_person_id`` is
    # bookkeeping, not delivery: a manager does not browse the auditor
    # console (that surface is gated on the auditor role) and may never
    # touch an MCP client. Without this notification and the "替下属复核"
    # tab it links to, the entire consequence is one database row nobody
    # ever sees — the failure this whole PR exists to fix, reproduced one
    # layer up.
    await _notify_followup_reviewers(
        db, followup=followup, trace_id=trace_id,
    )
    return followup


NO_RESPONSE_FOLLOWUP_TEMPLATE_KEY = "access_review.no_response_followup"


async def _notify_followup_reviewers(
    db: AsyncSession, *, followup: AccessCertification, trace_id: uuid.UUID | None,
) -> int:
    """One card per reviewer, listing how many reports they now own.

    Grouped per reviewer rather than per item: a manager with four
    non-responding reports would otherwise get four cards for what is one
    piece of work, which is how people learn to ignore a channel.
    """
    from app.notifications.fanout import enqueue_then_drain
    from app.notifications.service import (
        _DEFAULT_TEMPLATES,
        NotificationEnvelope,
    )

    rows = (await db.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == followup.id,
            AccessCertificationItem.assigned_reviewer_person_id.is_not(None),
        )
    )).scalars().all()
    if not rows:
        # Everyone's manager is unknown, so the items sit unassigned for
        # the auditor console to pick up. Nothing to notify, and saying
        # so in the log beats a silent zero.
        logger.info(
            "no_response_followup: %s has no assigned reviewers to notify",
            followup.id,
        )
        return 0

    per_reviewer: dict[uuid.UUID, list[Any]] = {}
    for row in rows:
        # The query already filters ``is_not(None)``; re-checked here so
        # the invariant is enforced rather than assumed by a reader (and
        # so the type checker can see it).
        reviewer_id = row.assigned_reviewer_person_id
        if reviewer_id is None:
            continue
        per_reviewer.setdefault(reviewer_id, []).append(row)

    tmpl = _DEFAULT_TEMPLATES.get(NO_RESPONSE_FOLLOWUP_TEMPLATE_KEY, {})
    reviewers = {
        p.id: p
        for p in (await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.id.in_(list(per_reviewer)),
                IdentityPerson.deleted_at.is_(None),
            )
        )).scalars().all()
    }

    # Build every envelope FIRST, send after the loop. Dispatching inline
    # inside a recipient loop makes each outbox INSERT hold an
    # ``identity_persons`` FK lock across the vendor call — with a few
    # hundred managers that is a long transaction sitting on the table
    # every login reads, which is how a notification fan-out takes the
    # login path down with it. ``enqueue_then_drain`` writes all the
    # durable rows, commits, and only then delivers, so a crash mid-drain
    # leaves the rest PENDING for the retry worker.
    envelopes: list[NotificationEnvelope] = []
    for reviewer_id, items in per_reviewer.items():
        reviewer = reviewers.get(reviewer_id)
        if reviewer is None:
            # The recorded manager is gone (deleted / purged). Left
            # unnotified rather than silently re-pointed at somebody
            # else: guessing a substitute reviewer is how an access
            # decision ends up made by a person nobody chose.
            logger.warning(
                "no_response_followup: reviewer %s not found; %s item(s) "
                "left assigned but unnotified", reviewer_id, len(items),
            )
            continue
        holders = sorted({
            (it.holder_display_name or it.holder_email or "?") for it in items
        })
        holder_name = "、".join(holders[:3]) + (
            f" 等 {len(holders)} 人" if len(holders) > 3 else ""
        )
        context = {
            "holder_name": holder_name,
            "item_count": len(items),
            "link": "/my/todos?tab=accessReviewAssigned",
        }
        body = str(tmpl.get("body_text") or "")
        envelopes.append(NotificationEnvelope(
            recipient=reviewer,
            title=str(tmpl.get("subject") or "需要你替下属确认访问权限"),
            body_text=(
                body.format(**context) if body
                else f"你的下属有 {len(items)} 项权限未确认,需要你判断。"
            ),
            template_key=NO_RESPONSE_FOLLOWUP_TEMPLATE_KEY,
            context=context,
            url="/my/todos?tab=accessReviewAssigned",
            related_resource_type="access_certifications",
            related_resource_id=followup.id,
        ))

    if not envelopes:
        return 0
    result = await enqueue_then_drain(
        db, envelopes, trace_id=str(trace_id or uuid.uuid4()),
        actor_user_id=None, label="no_response_followup",
    )
    return int(result.get("enqueued", 0))


async def complete_campaign(
    db: AsyncSession,
    *,
    certification_id: uuid.UUID,
    actor_person_id: uuid.UUID,
    trace_id: uuid.UUID | None = None,
) -> AccessCertification:
    cert = await db.get(AccessCertification, certification_id)
    if cert is None:
        raise NotFoundError("AccessCertification", str(certification_id))
    if cert.status != STATUS_IN_PROGRESS:
        raise AppException(
            code="CAMPAIGN_NOT_OPEN",
            message=f"campaign {cert.id} is {cert.status}",
            status_code=409,
        )
    pending = int((await db.execute(
        select(func.count()).select_from(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == certification_id,
            AccessCertificationItem.decision == DECISION_PENDING,
        )
    )).scalar_one())
    no_response = 0
    if pending > 0:
        if cert.campaign_type == CAMPAIGN_TYPE_SELF_ATTESTATION:
            # A self-attestation campaign snapshots EVERY active employee,
            # and non-responders always leave their items ``pending`` — an
            # auditor cannot disposition them (``decide_item`` refuses
            # self-attestation items). Blocking on pending here left every
            # half-yearly all-employee campaign permanently ``in_progress``.
            # Close over non-responders by stamping the remaining pending
            # items to the terminal ``no_response`` disposition (no human
            # decided — ``decided_by_person_id`` stays NULL).
            now = _naive_now()
            await db.execute(
                update(AccessCertificationItem)
                .where(
                    AccessCertificationItem.certification_id == certification_id,
                    AccessCertificationItem.decision == DECISION_PENDING,
                )
                .values(decision=DECISION_NO_RESPONSE, decided_at=now)
            )
            no_response = pending
            # The consequence. Opened AFTER the stamp so the closing
            # campaign's own ``no_response`` count stays the audit
            # evidence of what happened, and the follow-up is a separate,
            # separately-auditable campaign rather than a mutation of it.
            await _open_no_response_followup(
                db, cert=cert, actor_person_id=actor_person_id,
                trace_id=trace_id,
            )
        else:
            raise AppException(
                code="CAMPAIGN_HAS_PENDING",
                message=f"{pending} item(s) still pending decision",
                status_code=409,
            )
    cert.status = STATUS_COMPLETED
    cert.completed_at = _naive_now()
    await AuditService(db).log(
        trace_id=trace_id or uuid.uuid4(),
        actor_type="user",
        actor_user_id=actor_person_id,
        action="auditor.access_certification.complete",
        resource_type="access_certifications",
        resource_id=str(cert.id),
        status="success",
        detail={"decided_count": cert.decided_count,
                "revoked_count": cert.revoked_count,
                "no_response_count": no_response},
    )
    return cert


# ════════════════════════════════════════════════════════════════════
# Reads
# ════════════════════════════════════════════════════════════════════


def _campaign_view(c: AccessCertification) -> dict[str, Any]:
    return {
        "id": str(c.id),
        "label": c.label,
        "campaign_type": c.campaign_type,
        "scope_config": c.scope_config or {},
        "status": c.status,
        "started_at": _to_naive(c.started_at).isoformat() if c.started_at else None,
        "due_at": _to_naive(c.due_at).isoformat() if c.due_at else None,
        "completed_at": (
            _to_naive(c.completed_at).isoformat() if c.completed_at else None
        ),
        "snapshot_item_count": c.snapshot_item_count,
        "decided_count": c.decided_count,
        "revoked_count": c.revoked_count,
        "notes": c.notes,
    }


async def list_campaigns(db: AsyncSession) -> list[dict[str, Any]]:
    rows = (await db.execute(
        select(AccessCertification).order_by(
            AccessCertification.started_at.desc()
        )
    )).scalars().all()
    return [_campaign_view(c) for c in rows]


async def get_campaign(
    db: AsyncSession, certification_id: uuid.UUID,
) -> dict[str, Any] | None:
    cert = await db.get(AccessCertification, certification_id)
    if cert is None:
        return None
    return _campaign_view(cert)


async def list_items(
    db: AsyncSession,
    *,
    certification_id: uuid.UUID,
    decision: str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> tuple[list[dict[str, Any]], int]:
    base = select(AccessCertificationItem).where(
        AccessCertificationItem.certification_id == certification_id,
    )
    cq = select(func.count()).select_from(AccessCertificationItem).where(
        AccessCertificationItem.certification_id == certification_id,
    )
    if decision:
        base = base.where(AccessCertificationItem.decision == decision)
        cq = cq.where(AccessCertificationItem.decision == decision)
    base = base.order_by(
        AccessCertificationItem.holder_email,
        AccessCertificationItem.access_label,
    ).offset((page - 1) * page_size).limit(page_size)
    rows = (await db.execute(base)).scalars().all()
    total = int((await db.execute(cq)).scalar_one())
    items = [
        {
            "id": str(r.id),
            "person_id": str(r.person_id) if r.person_id else None,
            "holder_email": r.holder_email,
            "holder_display_name": r.holder_display_name,
            "access_type": r.access_type,
            "access_ref": r.access_ref,
            "access_label": r.access_label,
            "resource_name": r.resource_name,
            "assigned_reviewer_person_id": (
                str(r.assigned_reviewer_person_id)
                if r.assigned_reviewer_person_id else None
            ),
            "decision": r.decision,
            "decision_note": r.decision_note,
            "revocation_status": r.revocation_status,
            "decided_at": (
                _to_naive(r.decided_at).isoformat() if r.decided_at else None
            ),
        }
        for r in rows
    ]
    return items, total


async def latest_campaign_status(
    db: AsyncSession, *, campaign_type: str | None = None,
) -> dict[str, Any] | None:
    """Latest campaign (optionally filtered by type) for the compliance
    control evaluators. Both callers pass an explicit ``campaign_type``:
    ``access_certification`` → ``software_seat`` (general user-access
    review), ``privileged_access_review`` → ``privileged_role``. The
    unfiltered (``campaign_type=None``) form returns the most recent
    campaign of any type and is kept for ad-hoc/diagnostic use only — do
    NOT wire it to a control, or a quarterly privileged campaign will mask
    a missing user-access review."""
    q = select(AccessCertification).order_by(
        AccessCertification.started_at.desc()
    ).limit(1)
    if campaign_type is not None:
        q = select(AccessCertification).where(
            AccessCertification.campaign_type == campaign_type
        ).order_by(AccessCertification.started_at.desc()).limit(1)
    cert = (await db.execute(q)).scalar_one_or_none()
    if cert is None:
        return None
    return {
        "id": str(cert.id),
        "label": cert.label,
        "status": cert.status,
        "due_at": _to_naive(cert.due_at),
        "completed_at": _to_naive(cert.completed_at),
        "snapshot_item_count": cert.snapshot_item_count,
    }


async def auto_open_quarterly_privileged(
    db: AsyncSession, *, due_after_days: int = 30,
    trace_id: uuid.UUID | None = None,
) -> dict[str, Any]:
    """Idempotently open a ``privileged_role`` campaign for the current
    quarter (label ``YYYY-QN``). Replaces the old quarterly
    privileged-review cron; called by
    ``auditor.privileged_review_auto_open_quarterly``."""
    now = _naive_now()
    quarter = (now.month - 1) // 3 + 1
    label = f"{now.year}-Q{quarter} 特权访问复核"
    existing = (await db.execute(
        select(AccessCertification).where(
            AccessCertification.campaign_type == "privileged_role",
            AccessCertification.label == label,
        )
    )).scalar_one_or_none()
    if existing is not None:
        return {"skipped_reason": "already_exists",
                "label": label, "certification_id": str(existing.id)}
    cert = await create_campaign(
        db, label=label, campaign_type="privileged_role", scope_config={},
        due_at=datetime.now(UTC) + timedelta(days=max(1, int(due_after_days))),
        actor_person_id=None, trace_id=trace_id,
    )
    return {"label": label, "certification_id": str(cert.id),
            "snapshot_item_count": cert.snapshot_item_count}


async def auto_open_quarterly_identity_consumer(
    db: AsyncSession, *, due_after_days: int = 30,
    trace_id: uuid.UUID | None = None,
) -> dict[str, Any]:
    """Idempotently open an ``identity_consumer`` recert campaign for the
    current quarter. Called by
    ``auditor.identity_consumer_recert_auto_open_quarterly``."""
    now = _naive_now()
    quarter = (now.month - 1) // 3 + 1
    label = f"{now.year}-Q{quarter} 身份消费方复核"
    existing = (await db.execute(
        select(AccessCertification).where(
            AccessCertification.campaign_type == "identity_consumer",
            AccessCertification.label == label,
        )
    )).scalar_one_or_none()
    if existing is not None:
        return {"skipped_reason": "already_exists",
                "label": label, "certification_id": str(existing.id)}
    cert = await create_campaign(
        db, label=label, campaign_type="identity_consumer", scope_config={},
        due_at=datetime.now(UTC) + timedelta(days=max(1, int(due_after_days))),
        actor_person_id=None, trace_id=trace_id,
    )
    return {"label": label, "certification_id": str(cert.id),
            "snapshot_item_count": cert.snapshot_item_count}


async def lapse_disable_identity_consumers(
    db: AsyncSession, *, trace_id: str,
) -> dict[str, int]:
    """Auto-disable consumers whose ``identity_consumer`` recert item is
    still ``pending`` past the deadline + grace window — stops PII egress
    for un-recertified consumers (the enforcement teeth behind the
    recert campaign). Idempotent: the coordinator skips consumers already
    disabled / already lapse-flagged. A later ``keep`` re-enables. Pairs
    with ``escalate_overdue`` (which raises the compliance incident)."""
    from app.core.coordinators.access_certification_coordinator import (
        disable_identity_consumer_for_lapse,
    )

    cutoff = _naive_now() - timedelta(days=_ESCALATION_GRACE_DAYS)
    rows = (await db.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.access_type == "identity_consumer",
            AccessCertificationItem.decision == DECISION_PENDING,
            AccessCertificationItem.decided_at_deadline.is_not(None),
            AccessCertificationItem.decided_at_deadline <= cutoff,
        )
    )).scalars().all()
    disabled = 0
    for item in rows:
        if await disable_identity_consumer_for_lapse(
            db, item=item, trace_id=trace_id,
        ):
            disabled += 1
    if disabled:
        await db.flush()
    return {"scanned": len(rows), "disabled": disabled}



# ── Manager surface for the no-response follow-up (PR5) ──────────────


async def assigned_reviews_for(
    db: AsyncSession, *, reviewer_person_id: uuid.UUID,
) -> dict[str, Any]:
    """Pending items this person owns as the ASSIGNED reviewer.

    Self-scoped: being named on the item is the authorization, exactly
    like ``my_open_self_attestation``. This exists because the two
    existing decide paths (the compliance router and the MCP tool) are
    both gated on ``can_manage_security`` — an ordinary manager holds
    neither role, so before PR5 an item assigned to them was reachable
    by nobody.
    """
    rows = (await db.execute(
        select(AccessCertificationItem, AccessCertification)
        .join(
            AccessCertification,
            AccessCertification.id == AccessCertificationItem.certification_id,
        )
        .where(
            AccessCertificationItem.assigned_reviewer_person_id == reviewer_person_id,
            AccessCertificationItem.decision == DECISION_PENDING,
            AccessCertification.status == STATUS_IN_PROGRESS,
            # Same allowlist the decide endpoint enforces. Listing more
            # than it can decide would render rows that 403 when clicked
            # — and would surface other campaign types' holder data on a
            # surface with no admin role behind it.
            AccessCertification.campaign_type.in_(
                SELF_SERVICE_REVIEWABLE_CAMPAIGN_TYPES,
            ),
        )
    )).all()

    items: list[dict[str, Any]] = []
    for item, cert in rows:
        due = _to_naive(cert.due_at)
        items.append({
            "id": str(item.id),
            "certification_id": str(cert.id),
            "campaign_label": cert.label,
            "campaign_type": cert.campaign_type,
            "due_at": due.isoformat() if due else None,
            "days_left": (
                (due.date() - _naive_now().date()).days if due else None
            ),
            "holder_person_id": str(item.person_id) if item.person_id else None,
            "holder_email": item.holder_email,
            "holder_display_name": item.holder_display_name,
            "access_type": item.access_type,
            "access_label": item.access_label,
            "resource_name": item.resource_name,
            # Why this landed on a manager's desk at all. Without it the
            # card reads as an arbitrary chore rather than "your report
            # did not answer, so it is your call".
            "reason": (
                "no_response"
                if cert.campaign_type == CAMPAIGN_TYPE_NO_RESPONSE_FOLLOWUP
                else "assigned"
            ),
        })
    items.sort(key=lambda r: (r["holder_display_name"] or "", r["access_type"]))
    by_holder: dict[str, int] = {}
    for row in items:
        key = row["holder_display_name"] or row["holder_email"] or "?"
        by_holder[key] = by_holder.get(key, 0) + 1
    return {
        "items": items,
        "total": len(items),
        "holders": by_holder,
    }


async def decide_assigned_review(
    db: AsyncSession,
    *,
    item_id: uuid.UUID,
    decision: str,
    note: str | None,
    reviewer_person_id: uuid.UUID,
    trace_id: uuid.UUID | None = None,
) -> AccessCertificationItem:
    """Decide one item the caller is the assigned reviewer of.

    A thin authorization shell over :func:`decide_item` — the decision,
    the counter bump, the revocation enqueue and the audit row all stay
    in the one implementation. The only thing added here is the check
    that the caller is the named reviewer, which is what makes this
    endpoint safe to expose without an admin role.
    """
    item = await db.get(AccessCertificationItem, item_id)
    if item is None:
        raise NotFoundError("AccessCertificationItem", str(item_id))
    cert = await db.get(AccessCertification, item.certification_id)
    if cert is None or cert.campaign_type not in SELF_SERVICE_REVIEWABLE_CAMPAIGN_TYPES:
        # Campaign-type allowlist. Being named as reviewer is not enough:
        # some campaign types name the SUBJECT as their own reviewer, and
        # this endpoint has no admin role behind it. See the constant.
        raise AppException(
            code="NOT_YOUR_REVIEW",
            message="this item is not decided through the self-service path",
            status_code=403,
        )
    if item.person_id is not None and item.person_id == reviewer_person_id:
        # Self-review is not a review. Also closes the self-managed
        # holder case (``manager_person_id == holder.id``), where the
        # follow-up would otherwise hand somebody their own non-response
        # to rule on.
        raise AppException(
            code="NOT_YOUR_REVIEW",
            message="you cannot decide your own access here",
            status_code=403,
        )
    if item.assigned_reviewer_person_id != reviewer_person_id:
        # Deliberately the same 403 shape whether the item exists and
        # belongs to somebody else or the caller simply is not its
        # reviewer: a manager probing ids should not learn who owns what.
        raise AppException(
            code="NOT_YOUR_REVIEW",
            message="this item is not assigned to you",
            status_code=403,
        )
    return await decide_item(
        db, item_id=item_id, decision=decision, note=note,
        actor_person_id=reviewer_person_id, trace_id=trace_id,
    )


async def escalate_overdue(
    db: AsyncSession, *, trace_id: str,
) -> dict[str, int]:
    """14-day past-deadline escalation sweep over pending cert items:
    set ``escalated_at`` + write an ``access_review_overdue`` compliance
    incident (idempotent — once ``escalated_at`` is set the row is
    skipped). Generalized from the old privileged-review escalation."""
    from app.compliance.models import (
        INCIDENT_OPEN,
        SEVERITY_WARNING,
        TYPE_ACCESS_REVIEW_OVERDUE,
        ComplianceIncident,
    )

    cutoff = _naive_now() - timedelta(days=_ESCALATION_GRACE_DAYS)
    rows = (await db.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.decision == DECISION_PENDING,
            AccessCertificationItem.decided_at_deadline.is_not(None),
            AccessCertificationItem.decided_at_deadline <= cutoff,
            AccessCertificationItem.escalated_at.is_(None),
        )
    )).scalars().all()
    # Which of these belong to a self_attestation campaign? Those also get a
    # Feishu/inbox nudge to the holder + their manager (batched per holder),
    # on top of the generic compliance incident every type already raises.
    self_att_cert_ids = {
        cid for (cid,) in (await db.execute(
            select(AccessCertification.id).where(
                AccessCertification.id.in_({r.certification_id for r in rows}),
                AccessCertification.campaign_type
                == CAMPAIGN_TYPE_SELF_ATTESTATION,
            )
        )).all()
    } if rows else set()
    self_att_overdue: list[AccessCertificationItem] = []
    escalated = 0
    escalated_item_ids: list[tuple[str, str]] = []
    for item in rows:
        item.escalated_at = _naive_now()
        is_self_att = bool(
            item.certification_id in self_att_cert_ids and item.person_id
        )
        if is_self_att:
            # Self-attestation incidents are raised ONCE per holder below, not
            # one-per-item — a non-responder holding 15 entitlements must not
            # spawn 15 ComplianceIncident rows (the incident storm at the
            # ≈500-employee half-yearly scale).
            self_att_overdue.append(item)
        else:
            db.add(ComplianceIncident(
                incident_type=TYPE_ACCESS_REVIEW_OVERDUE,
                severity=SEVERITY_WARNING,
                status=INCIDENT_OPEN,
                person_id=item.assigned_reviewer_person_id,
                resource_type="access_certification_item",
                resource_id=item.id,
                resource_label=(
                    f"{item.access_label} — "
                    f"{item.holder_email or item.access_ref}"
                ),
                title=(
                    f"Access review overdue: {item.access_label} pending "
                    f"> 14 days past deadline"
                ),
                detail={
                    "certification_id": str(item.certification_id),
                    "item_id": str(item.id),
                    "assigned_reviewer_person_id": (
                        str(item.assigned_reviewer_person_id)
                        if item.assigned_reviewer_person_id else None
                    ),
                    "decided_at_deadline": (
                        _to_naive(item.decided_at_deadline).isoformat()
                        if item.decided_at_deadline else None
                    ),
                },
                detected_at=_naive_now(),
            ))
        escalated += 1
        escalated_item_ids.append(
            (str(item.id), str(item.certification_id)),
        )
    # ONE audit row for the sweep, listing the escalated items — not one
    # row (and one ``AuditService`` instantiation) per item (#2188-8).
    # N overdue items meant N INSERTs inside the transaction this sweep
    # already holds locks in, lengthening exactly the window #2143 was
    # about. The per-item ids stay in ``detail`` so the trail is intact.
    if escalated_item_ids:
        await AuditService(db).log(
            trace_id=trace_id,
            actor_type="system",
            action="auditor.access_certification.escalated",
            status="success",
            resource_type="access_certification_item",
            resource_id=escalated_item_ids[0][0],
            detail={
                "escalated_count": len(escalated_item_ids),
                "items": [
                    {"item_id": iid, "certification_id": cid}
                    for iid, cid in escalated_item_ids[:500]
                ],
                "items_truncated": len(escalated_item_ids) > 500,
            },
        )

    # Self-attestation: one incident per (holder, campaign), summarising all
    # of that holder's overdue items. Created in the same pass that stamps
    # ``escalated_at`` so re-runs (filtered on ``escalated_at IS NULL``) never
    # raise a duplicate.
    by_holder_camp: dict[tuple[uuid.UUID, uuid.UUID], list[AccessCertificationItem]] = {}
    for it in self_att_overdue:
        by_holder_camp.setdefault((it.person_id, it.certification_id), []).append(it)
    for (pid, cid), its in by_holder_camp.items():
        db.add(ComplianceIncident(
            incident_type=TYPE_ACCESS_REVIEW_OVERDUE,
            severity=SEVERITY_WARNING,
            status=INCIDENT_OPEN,
            person_id=pid,
            resource_type="access_certification",
            resource_id=cid,
            resource_label=(
                f"{its[0].holder_email or pid} — "
                f"{len(its)} 项访问自查逾期"
            ),
            title=(
                f"Access self-attestation overdue: {len(its)} item(s) pending "
                f"> 14 days past deadline"
            ),
            detail={
                "certification_id": str(cid),
                "person_id": str(pid),
                "item_count": len(its),
                "item_ids": [str(i.id) for i in its[:50]],
            },
            detected_at=_naive_now(),
        ))
    if escalated:
        # COMMIT the escalation state before notifying (#2143). The
        # fan-out below commits per recipient anyway, so leaving the
        # sweep's own writes uncommitted here would just make the first
        # of those commits publish them at an arbitrary point — better
        # to make the state durable deliberately, and to release the
        # identity_persons FK locks the incident INSERTs took before any
        # external call happens.
        await db.commit()
    nudged = await _nudge_self_attestation_overdue(
        db, items=self_att_overdue, trace_id=trace_id,
    )
    return {"escalated": escalated, "scanned": len(rows), "nudged": nudged}


async def _nudge_self_attestation_overdue(
    db: AsyncSession, *, items: list[AccessCertificationItem], trace_id: str,
) -> int:
    """One Feishu + inbox nudge per holder (and their manager) for
    self-attestation items that just crossed the grace window — never
    one-per-item. Best-effort: a send hiccup must not fail the sweep."""
    if not items:
        return 0
    from app.notifications.service import (
        _SEE_TEMPLATE_SENTINEL,
        CardAction,
        LocalizedText,
        NotificationEnvelope,
    )

    by_holder: dict[uuid.UUID, int] = {}
    for it in items:
        by_holder[it.person_id] = by_holder.get(it.person_id, 0) + 1

    holders = {
        p.id: p for p in (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(by_holder.keys()))
        )).scalars().all()
    }
    mgr_ids = {
        p.manager_person_id for p in holders.values()
        if p.manager_person_id is not None
    }
    managers = {
        p.id: p for p in (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(mgr_ids))
        )).scalars().all()
    } if mgr_ids else {}

    # Collected, then fanned out with per-row commits AFTER the caller's
    # state changes are durable. The sweep that calls this first UPDATEs
    # up to thousands of items and INSERTs ~100 compliance_incidents
    # (each FK → identity_persons), flushes, and used to hold every one
    # of those locks across ~300 serial Feishu/email sends (#2143) —
    # blocking identity_persons writes, i.e. logins, for the duration.
    envelopes: list[NotificationEnvelope] = []
    nudged = 0
    for pid, n in by_holder.items():
        holder = holders.get(pid)
        if holder is None or holder.work_email is None:
            continue
        envelopes.append(NotificationEnvelope(
            recipient=holder,
            title="访问自查已逾期,请尽快完成",
            # Sentinel rather than the rendered Chinese: a stored non-sentinel
            # body short-circuits ``render_inbox_body``, so the bell would keep
            # showing Chinese to an EN/JA reader whose Feishu card and email
            # already render the localised template.
            body_text=_SEE_TEMPLATE_SENTINEL,
            template_key=SELF_ATTESTATION_OVERDUE_TEMPLATE_KEY,
            channels=["feishu", "email"],
            url="/me/access-review",
            context={"pending_count": n},
            card_actions=[CardAction(label=LocalizedText(zh="去确认",
                                                        en="Review now",
                                                        ja="確認する"),
                                     url="/me/access-review",
                                     style="primary")],
        ))
        nudged += 1
        # Manager gets a heads-up too (no action button — it's their report's).
        mgr = managers.get(holder.manager_person_id) if holder.manager_person_id else None
        if mgr is not None and mgr.work_email:
            envelopes.append(NotificationEnvelope(
                recipient=mgr,
                title="下属访问自查逾期",
                body_text=_SEE_TEMPLATE_SENTINEL,
                template_key=SELF_ATTESTATION_OVERDUE_MANAGER_TEMPLATE_KEY,
                channels=["feishu"],
                context={
                    "holder_name": holder.display_name,
                    "pending_count": n,
                },
            ))

    from app.notifications import fanout
    await fanout.enqueue_then_drain(
        db, envelopes, trace_id=str(trace_id or ""),
        label="auditor.self_attestation.overdue_nudge",
    )
    return nudged


# ════════════════════════════════════════════════════════════════════
# Self-attestation (员工半年度访问自查) — holder-facing batch path
# ════════════════════════════════════════════════════════════════════


def _self_item_sort_key(it: AccessCertificationItem) -> tuple:
    raw = it.raw_payload or {}
    elevated = 0 if raw.get("sensitivity") == "elevated" else 1
    return (elevated, it.access_type, it.access_label or "")


def _self_item_view(it: AccessCertificationItem) -> dict[str, Any]:
    raw = it.raw_payload or {}
    return {
        "id": str(it.id),
        "access_type": it.access_type,
        "access_label": it.access_label,
        "resource_name": it.resource_name,
        "sensitivity": raw.get("sensitivity") or "standard",
        "removable": bool(raw.get("removable")),
        "granted_at": raw.get("granted_at"),
        "last_used_at": raw.get("last_used_at"),
        "decision": it.decision,
        "revocation_status": it.revocation_status,
        "decided_at": (
            _to_naive(it.decided_at).isoformat() if it.decided_at else None
        ),
    }


async def latest_open_self_attestation(
    db: AsyncSession,
) -> AccessCertification | None:
    return (await db.execute(
        select(AccessCertification).where(
            AccessCertification.campaign_type == CAMPAIGN_TYPE_SELF_ATTESTATION,
            AccessCertification.status == STATUS_IN_PROGRESS,
        ).order_by(AccessCertification.started_at.desc()).limit(1)
    )).scalar_one_or_none()


async def my_open_self_attestation(
    db: AsyncSession, *, person_id: uuid.UUID,
) -> dict[str, Any]:
    """The currently-open self-attestation campaign as the holder sees it:
    summary + their own items (sorted elevated-first). ``open`` is False
    when no campaign is in progress — the employee surface only appears
    then. Pending vs decided are both returned so the page can show
    progress and let a holder revisit before the campaign closes."""
    cert = await latest_open_self_attestation(db)
    if cert is None:
        return {"open": False, "campaign": None, "items": [],
                "pending_count": 0, "total_count": 0}
    items = (await db.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert.id,
            AccessCertificationItem.person_id == person_id,
        )
    )).scalars().all()
    items.sort(key=_self_item_sort_key)
    pending = sum(1 for it in items if it.decision == DECISION_PENDING)
    due = _to_naive(cert.due_at)
    days_left = (due.date() - _naive_now().date()).days if due else None
    return {
        "open": True,
        "campaign": {
            "id": str(cert.id),
            "label": cert.label,
            "due_at": due.isoformat() if due else None,
            "days_left": days_left,
        },
        "items": [_self_item_view(it) for it in items],
        "pending_count": pending,
        "total_count": len(items),
        # PR5: the model's suggestion, beside the holder's decision and
        # never in it. ``None`` when the feature is off, when advice has
        # not been generated for this holder yet, or when generation
        # failed — the page renders the reason instead of a blank space
        # that would read as "nothing to worry about".
        "advice": await _self_attestation_advice_view(
            db, certification_id=cert.id, person_id=person_id,
        ),
    }


async def _self_attestation_advice_view(
    db: AsyncSession, *, certification_id: uuid.UUID, person_id: uuid.UUID,
) -> dict[str, Any] | None:
    """The holder's advice row, shaped for the review page."""
    from app.domains.auditor.models import AuditorSelfAttestationAdvice
    from app.domains.auditor.self_attestation_ai import advice_enabled

    if not await advice_enabled(db):
        return None
    row = (await db.execute(
        select(AuditorSelfAttestationAdvice).where(
            AuditorSelfAttestationAdvice.certification_id == certification_id,
            AuditorSelfAttestationAdvice.person_id == person_id,
        )
    )).scalars().first()
    if row is None:
        # Generated nightly in batches, so an early-opening holder can
        # legitimately arrive before theirs exists. Reported as "not yet"
        # rather than as a failure.
        return {"status": "pending", "failure_code": None}
    if row.failure_code:
        return {"status": "unavailable", "failure_code": row.failure_code}
    return {
        "status": "ready",
        "failure_code": None,
        "summary_zh": row.summary_zh,
        "summary_en": row.summary_en,
        "type_recommendations": row.type_recommendations or {},
        # The fact package the narrative came from. Returned so a holder
        # who disagrees can see what it was told — including how many of
        # their items Helix could not measure at all.
        "facts": row.facts_snapshot or {},
        "generated_at": (
            row.generated_at.isoformat() if row.generated_at else None
        ),
    }


async def submit_self_attestation(
    db: AsyncSession,
    *,
    actor_person_id: uuid.UUID,
    keep_item_ids: list[uuid.UUID],
    remove_item_ids: list[uuid.UUID],
    remove_reason: str | None = None,
    trace_id: uuid.UUID | None = None,
) -> dict[str, Any]:
    """Batch holder decision over the holder's OWN self-attestation items.
    ``keep`` = "still need it" (the attestation evidence); ``remove`` =
    "我不需要了" → records revoke + routes to the access type's existing
    self-service removal via the coordinator. Idempotent: already-decided
    items are skipped (re-submitting a partially-done review is safe).
    Every item must belong to the actor and to an OPEN self_attestation
    campaign, else the whole batch is rejected before any write."""
    keep_set = {uuid.UUID(str(x)) for x in keep_item_ids}
    remove_set = {uuid.UUID(str(x)) for x in remove_item_ids}
    overlap = keep_set & remove_set
    if overlap:
        raise AppException(
            code="AMBIGUOUS_DECISION",
            message=f"item(s) in both keep and remove: {sorted(map(str, overlap))}",
            status_code=422,
        )
    all_ids = keep_set | remove_set
    if not all_ids:
        return {"kept": 0, "removed": 0, "skipped": 0}

    # FOR UPDATE, ordered by id: the ``decision != PENDING`` skip below is
    # a read-then-write feeding an unconditional counter delta, so a
    # double-submit (two tabs, or a retried request) used to have both
    # copies pass the guard and count the same item twice. Deterministic
    # lock order keeps two concurrent batches that overlap from
    # deadlocking each other.
    items = (await db.execute(
        select(AccessCertificationItem)
        .where(AccessCertificationItem.id.in_(all_ids))
        .order_by(AccessCertificationItem.id)
        .with_for_update()
    )).scalars().all()
    by_id = {it.id: it for it in items}
    missing = all_ids - set(by_id)
    if missing:
        raise NotFoundError(
            "AccessCertificationItem", str(sorted(map(str, missing))[0]),
        )

    # Validate ownership + campaign-open BEFORE any mutation.
    cert_cache: dict[uuid.UUID, AccessCertification] = {}
    for iid in all_ids:
        it = by_id[iid]
        if it.person_id != actor_person_id:
            raise AppException(
                code="NOT_YOUR_ACCESS",
                message="you can only attest your own access items",
                status_code=403,
            )
        cert = cert_cache.get(it.certification_id)
        if cert is None:
            cert = await db.get(AccessCertification, it.certification_id)
            cert_cache[it.certification_id] = cert
        if cert is None or cert.campaign_type != CAMPAIGN_TYPE_SELF_ATTESTATION:
            raise AppException(
                code="NOT_SELF_ATTESTATION",
                message="item is not part of a self-attestation campaign",
                status_code=409,
            )
        if cert.status != STATUS_IN_PROGRESS:
            raise AppException(
                code="CAMPAIGN_NOT_OPEN",
                message=f"campaign {cert.id} is {cert.status}",
                status_code=409,
            )

    from app.core.coordinators.access_certification_coordinator import (
        enqueue_self_service_removal_for_cert_item,
    )

    now = _naive_now()
    kept = removed = skipped = failed = 0
    # Accumulate per-campaign counter deltas and apply them as ONE atomic
    # SQL increment after the loop, so concurrent submissions by different
    # employees on the same campaign can't lose-update the shared counters
    # (a half-yearly campaign has hundreds of employees submitting at once).
    counter_deltas: dict[uuid.UUID, list[int]] = {}
    for iid in all_ids:
        it = by_id[iid]
        if it.decision != DECISION_PENDING:
            skipped += 1
            continue
        cert = cert_cache[it.certification_id]
        if iid in remove_set:
            # The removal carries the external/flush side effects (a cloud
            # grant revoke calls the provider + flushes the Helix flip), so
            # run ONLY it inside a per-item SAVEPOINT — one item's failure
            # rolls back just that item and never poisons the batch txn nor
            # reverts an earlier item's applied revoke (the split-brain the
            # unisolated loop risked). On failure expire the item so the
            # rolled-back in-memory mutations aren't flushed on outer commit;
            # the decision + counters are recorded ONLY after the savepoint
            # succeeds (a SAVEPOINT rollback does NOT restore Python attrs),
            # so a failed item stays pending and is re-submittable.
            try:
                async with db.begin_nested():
                    await enqueue_self_service_removal_for_cert_item(
                        db, item=it, actor_user_id=actor_person_id,
                        reason=remove_reason or "员工自查：我不需要了",
                        trace_id=str(trace_id or uuid.uuid4()),
                    )
            except OperationalError:
                # A deadlock / lock-wait timeout poisons the whole session, so
                # every later item in the batch would also fail. Re-raise so
                # the request fails cleanly and the user retries — rather than
                # silently stranding the remainder of the batch as pending.
                raise
            except Exception:  # noqa: BLE001
                logger.exception(
                    "self-attestation removal failed for item %s; left "
                    "pending", iid,
                )
                db.expire(it)
                failed += 1
                continue
            it.decision = DECISION_REVOKE
            it.decided_by_person_id = actor_person_id
            it.decided_at = now
            it.decision_note = remove_reason
            cd = counter_deltas.setdefault(it.certification_id, [0, 0])
            cd[0] += 1
            cd[1] += 1
            removed += 1
        else:
            # Keep is a pure attribute write (no side effect) — no savepoint.
            it.decision = DECISION_KEEP
            it.decided_by_person_id = actor_person_id
            it.decided_at = now
            counter_deltas.setdefault(it.certification_id, [0, 0])[0] += 1
            kept += 1

    for cid, (dd, rd) in counter_deltas.items():
        if dd:
            await db.execute(
                update(AccessCertification)
                .where(AccessCertification.id == cid)
                .values(
                    decided_count=AccessCertification.decided_count + dd,
                    revoked_count=AccessCertification.revoked_count + rd,
                )
            )
    # Audit status must not read ``success`` when access removals failed —
    # this is compliance evidence, and an auditor scanning by status would
    # otherwise miss that some "我不需要了" revocations never happened.
    # ``failed`` (nothing removed but failures) / ``partial`` (some removed,
    # some failed) / ``success`` (no failures).
    if failed > 0:
        audit_status = "failed" if removed == 0 else "partial"
    else:
        audit_status = "success"
    await AuditService(db).log(
        trace_id=trace_id or uuid.uuid4(),
        actor_type="user",
        actor_user_id=actor_person_id,
        action="auditor.access_certification.self_attest",
        resource_type="access_certifications",
        resource_id=str(next(iter(cert_cache))) if cert_cache else None,
        status=audit_status,
        detail={"kept": kept, "removed": removed, "skipped": skipped,
                "failed": failed},
    )
    return {"kept": kept, "removed": removed, "skipped": skipped,
            "failed": failed}


async def notify_self_attestation_holders(
    db: AsyncSession, *, certification: AccessCertification,
    trace_id: uuid.UUID | None = None,
) -> dict[str, int]:
    """One batched card + inbox row per holder with ≥1 pending item —
    never one-per-item (the explicit anti-noise requirement).

    **Idempotent by explicit check, not by hope.** The old docstring
    claimed "the outbox dedups on (recipient, template_key,
    related_resource_id)" — it does not; ``enqueue`` inserts a fresh row
    every call. The skip below is what actually makes a re-run safe, and
    it is what lets this double as the top-up path: if the process dies
    part-way through the fan-out, calling it again notifies exactly the
    holders who have no card yet.
    """
    from collections import defaultdict

    from app.notifications.service import (
        _SEE_TEMPLATE_SENTINEL,
        CardAction,
        CardStat,
        LocalizedText,
        NotificationEnvelope,
    )

    items = (await db.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == certification.id,
            AccessCertificationItem.decision == DECISION_PENDING,
            AccessCertificationItem.person_id.is_not(None),
        )
    )).scalars().all()
    by_person: dict[uuid.UUID, list[AccessCertificationItem]] = defaultdict(list)
    for it in items:
        by_person[it.person_id].append(it)
    if not by_person:
        return {"notified": 0}

    persons = {
        p.id: p for p in (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(by_person.keys()))
        )).scalars().all()
    }
    # Holders who already have a card for THIS campaign. Skipping them is
    # what makes a re-run (or a resumed fan-out) safe — see the docstring.
    from app.notifications.models import NotificationOutbox
    already = {
        pid for (pid,) in (await db.execute(
            select(NotificationOutbox.recipient_person_id).where(
                NotificationOutbox.template_key
                == SELF_ATTESTATION_TEMPLATE_KEY,
                # Compared as a STRING: the column is a raw CHAR(36)
                # (only annotated as UUID), so rows come back — and match
                # — as dashed text. Binding a UUID object here silently
                # matched nothing. Same family as
                # [[reference_scheduled_jobs_nodash_uuid_bug]].
                NotificationOutbox.related_resource_id == str(certification.id),
            )
        )).all() if pid is not None
    }

    due = _to_naive(certification.due_at)
    due_label = due.strftime("%Y-%m-%d") if due else ""
    # Envelopes are COLLECTED here and fanned out after the loop by
    # ``enqueue_then_drain``, which commits per row. Sending inline used
    # to keep every holder's ``identity_persons`` row FK-locked (via its
    # outbox INSERT) across the whole ~500-recipient × 2-channel fan-out
    # — plus the thousands of freshly inserted certification items, each
    # holding the same parent rows — so any concurrent login's
    # ``UPDATE identity_persons`` queued until 1205 (#2128, same shape as
    # the 2026-06 outage). It also outran the 600s Redis open-lock,
    # letting a cron tick and an admin "open now" double-open the very
    # campaign that lock exists to serialise.
    envelopes: list[NotificationEnvelope] = []
    for pid, its in by_person.items():
        person = persons.get(pid)
        if person is None or not person.work_email:
            continue
        if pid in already:
            continue
        total = len(its)
        elevated = sum(
            1 for it in its
            if (it.raw_payload or {}).get("sensitivity") == "elevated"
        )
        env = NotificationEnvelope(
            recipient=person,
            title="半年度访问自查：请确认你的访问权限",
            # Sentinel, not the rendered Chinese: ``render_inbox_body`` returns
            # a stored non-sentinel body verbatim, so the bell would keep
            # showing Chinese to a reader whose card and email already render
            # the localised template.
            body_text=_SEE_TEMPLATE_SENTINEL,
            template_key=SELF_ATTESTATION_TEMPLATE_KEY,
            channels=["feishu", "email"],
            url="/me/access-review",
            context={
                "total": total,
                "elevated": elevated,
                "due_label": due_label,
            },
            related_resource_type="access_certifications",
            related_resource_id=certification.id,
            card_subtitle=LocalizedText(
                zh="每半年一次的访问权限确认 · 安全合规",
                en="Half-yearly access review · Security & compliance",
                ja="半期ごとのアクセス権限確認 · セキュリティコンプライアンス",
            ),
            card_stats=[
                CardStat(
                    label=LocalizedText(
                        zh="待确认访问",
                        en="Access to review",
                        ja="確認待ちのアクセス権",
                    ),
                    value=LocalizedText(
                        zh=f"{total} 项",
                        en=f"{total} {'item' if total == 1 else 'items'}",
                        ja=f"{total} 件",
                    ),
                ),
                CardStat(
                    label=LocalizedText(
                        zh="高敏感",
                        en="Highly sensitive",
                        ja="高機密",
                    ),
                    value=LocalizedText(
                        zh=f"{elevated} 项",
                        en=f"{elevated} {'item' if elevated == 1 else 'items'}",
                        ja=f"{elevated} 件",
                    ),
                ),
            ],
            card_actions=[
                CardAction(label=LocalizedText(zh="去确认",
                                               en="Review now",
                                               ja="確認する"),
                           url="/me/access-review",
                           style="primary"),
            ],
        )
        envelopes.append(env)

    from app.notifications import fanout
    result = await fanout.enqueue_then_drain(
        db, envelopes, trace_id=str(trace_id or ""),
        label="auditor.self_attestation.open",
    )
    # ``notified`` keeps meaning "holders who have a durable card record",
    # which is what the caller reports and what makes a re-run safe — a
    # row that failed to DELIVER is still queued for the retry worker.
    return {"notified": result["enqueued"], "delivered": result["delivered"]}


def _self_attestation_label(period_year: int, half: str) -> str:
    return f"{period_year}-{half} 全员访问自查自证"


async def open_self_attestation_campaign(
    db: AsyncSession, *, period_year: int, half: str,
    due_after_days: int = 21, actor_person_id: uuid.UUID | None = None,
    trace_id: uuid.UUID | None = None,
) -> dict[str, Any]:
    """Open the self-attestation campaign for one period (idempotent by
    label). Snapshots every active employee's footprint and fires one
    batched card per holder. Shared by the cron and the admin "open now"
    action.

    A Redis advisory lock keyed by the label serialises the read-then-create
    so the daily cron tick and an admin "open now" can't both pass the
    existence check and double-open the campaign (which would snapshot twice
    and blast every employee two cards). A busy lock means another opener is
    mid-flight → report ``open_in_progress`` (no-op)."""
    from app.core.distributed_lock import LockBusy, distributed_lock

    label = _self_attestation_label(period_year, half)
    try:
        # TTL must comfortably outlast a full-footprint snapshot of every
        # active employee (≈500 × entitlements) + the batched-card fan-out,
        # so the lock can't expire mid-open and let a second opener in.
        async with distributed_lock(f"self_attest_open:{label}", ttl=600):
            existing = (await db.execute(
                select(AccessCertification).where(
                    AccessCertification.campaign_type
                    == CAMPAIGN_TYPE_SELF_ATTESTATION,
                    AccessCertification.label == label,
                )
            )).scalar_one_or_none()
            if existing is not None:
                if existing.status != STATUS_IN_PROGRESS:
                    return {"skipped_reason": "already_exists",
                            "label": label,
                            "certification_id": str(existing.id)}
                # Still open → the fan-out below runs as a TOP-UP.
                # A crash part-way through the original fan-out would
                # otherwise leave those holders with no card and no
                # recovery path: re-opening is refused by this very
                # check, and the outbox retry worker only backstops rows
                # that EXIST. The notify helper skips holders who
                # already have one, so this is a no-op on a healthy
                # campaign — one query per daily tick.
                #
                # It is recorded and returned BELOW, outside the lock —
                # the whole point of ``_finish_self_attestation_open`` is
                # that the fan-out is not held inside a 600s TTL, and the
                # top-up is precisely the case where it can be large
                # (that is what "the original fan-out crashed" means).
                cert, topped_up = existing, True
            else:
                cert = await create_campaign(
                    db, label=label,
                    campaign_type=CAMPAIGN_TYPE_SELF_ATTESTATION,
                    scope_config={"period": f"{period_year}-{half}"},
                    due_at=datetime.now(UTC) + timedelta(
                        days=max(1, int(due_after_days))),
                    actor_person_id=actor_person_id, trace_id=trace_id,
                    # The one sanctioned caller: we hold the open lock, we
                    # checked the label, and we notify holders below.
                    _allow_self_attestation=True,
                )
                topped_up = False
                # Commit INSIDE the lock so the new campaign row is durably
                # visible before the lock is released. The handler used to
                # commit only after this function returned (i.e. after the
                # lock was gone), leaving a window where a racing opener
                # (admin "open now" overlapping the daily cron) could pass
                # the existence check and double-snapshot + double-card
                # every employee.
                await db.commit()
    except LockBusy:
        return {"skipped_reason": "open_in_progress", "label": label}

    return await _finish_self_attestation_open(
        db, cert=cert, label=label, trace_id=trace_id, topped_up=topped_up,
    )


async def _finish_self_attestation_open(
    db: AsyncSession, *, cert: AccessCertification, label: str,
    trace_id: uuid.UUID | None, topped_up: bool,
) -> dict[str, Any]:
    """Fan out the campaign's cards. Runs OUTSIDE the open lock.

    Split out so the fresh-open and top-up paths cannot drift.
    """
    # Fan out AFTER the lock is released and the campaign row is durable
    # (#2128). Two problems this ordering fixes:
    #
    # * the fan-out used to run inside the lock and inside the same
    #   uncommitted transaction as thousands of item INSERTs, so ~500
    #   ``identity_persons`` rows stayed FK-locked across ~1000 external
    #   sends — the login brown-out shape;
    # * ~500 × 2 channel sends can plausibly exceed the lock's 600s TTL,
    #   at which point the lock silently expires and a racing opener does
    #   exactly the double-open the lock exists to prevent.
    #
    # Releasing the OPEN lock first is safe for the open itself — the
    # committed campaign row is the mutual exclusion from there on. It is
    # NOT sufficient for the fan-out (#2242): the only dedup is the
    # ``already`` set read at the top of ``notify_self_attestation_holders``,
    # so two openers that both get past the existence check (the daily
    # 08:00 cron and an admin "open now") each read ``already`` before the
    # other's phase-1 enqueue lands and each enqueue a card for the SAME
    # holders — every one of them gets two. The window is only the
    # phase-1 enqueue on a fresh open, but the TOP-UP path (whose
    # precondition is "the previous fan-out crashed", i.e. many uncarded
    # holders) is both the slowest and the day an admin is most likely to
    # click "open now" next to the cron.
    #
    # A SEPARATE lock, so the fix does not undo #2128 (the fan-out must
    # not run inside the open lock, whose TTL it can outlive). TTL is
    # sized for ~500 holders × 2 channels with retries; a busy lock means
    # another fan-out for this same label is in flight, which is exactly
    # the case we want to skip rather than duplicate.
    from app.core.distributed_lock import LockBusy, distributed_lock

    fanout_skipped = False
    try:
        async with distributed_lock(f"self_attest_fanout:{label}", ttl=1800):
            notify = await notify_self_attestation_holders(
                db, certification=cert, trace_id=trace_id,
            )
    except LockBusy:
        fanout_skipped = True
        logger.info(
            "self-attestation fan-out for %s skipped — another opener is "
            "already fanning out this campaign (#2242)", label,
        )
        notify = {"notified": 0, "delivered": 0}
    return {"label": label, "certification_id": str(cert.id),
            "snapshot_item_count": cert.snapshot_item_count,
            "notified": notify.get("notified", 0),
            "fanout_skipped": fanout_skipped,
            "topped_up": topped_up}


async def auto_open_semiannual_self_attestation(
    db: AsyncSession, *, due_after_days: int = 21, window_days: int = 7,
    trace_id: uuid.UUID | None = None,
) -> dict[str, Any]:
    """Daily cron: open the half-yearly campaign ONLY inside the open
    window just after a boundary — ``[Jun 1, Jun 1+window)`` for H1 and
    ``[Dec 1, Dec 1+window)`` for H2. This deliberately does NOT
    retroactively fire for a boundary that passed before this feature
    existed (a mid-cycle deploy must not blast every employee on its
    first run — the 存量风暴 guard); the next real boundary opens it
    automatically, and an admin can open the current period on demand via
    ``open_self_attestation_campaign``. Idempotent by label within the
    window."""
    now = _naive_now()
    y = now.year
    if now.month == 6 and 1 <= now.day < 1 + max(1, window_days):
        period_year, half = y, "H1"
    elif now.month == 12 and 1 <= now.day < 1 + max(1, window_days):
        period_year, half = y, "H2"
    else:
        return {"skipped_reason": "not_in_open_window",
                "date": now.date().isoformat()}
    return await open_self_attestation_campaign(
        db, period_year=period_year, half=half,
        due_after_days=due_after_days, trace_id=trace_id,
    )


# ════════════════════════════════════════════════════════════════════
# Governance read — auditor coverage / flagged-removal queue / output
# ════════════════════════════════════════════════════════════════════


async def build_self_attestation_summary(
    db: AsyncSession, *, certification_id: uuid.UUID,
) -> dict[str, Any] | None:
    """Auditor-facing roll-up of a self-attestation campaign: coverage,
    per-decision + per-type counts, the flagged-removal disposition queue
    (decision=revoke with its enforcement status), and non-responders
    (holders with still-pending items, overdue flagged). Read-only; feeds
    the auditor console, the governance landing tile, and the Word/PDF
    export. Returns None when the campaign id isn't a self_attestation
    campaign."""
    cert = await db.get(AccessCertification, certification_id)
    if cert is None or cert.campaign_type != CAMPAIGN_TYPE_SELF_ATTESTATION:
        return None

    # Column projection, not whole ORM rows (#2188-7). A campaign is
    # thousands of items and each carries a ``raw_payload`` JSON blob
    # that nothing below reads — and this summary hangs off the
    # governance landing page, so it is paid on every load. The list is
    # exactly what the loop touches; a new ``it.x`` read needs a new
    # entry here, which fails loudly rather than returning None.
    items = (await db.execute(
        select(
            AccessCertificationItem.id,
            AccessCertificationItem.person_id,
            AccessCertificationItem.decision,
            AccessCertificationItem.access_type,
            AccessCertificationItem.access_label,
            AccessCertificationItem.holder_email,
            AccessCertificationItem.holder_display_name,
            AccessCertificationItem.revocation_status,
            AccessCertificationItem.decided_at,
            AccessCertificationItem.decided_at_deadline,
        ).where(
            AccessCertificationItem.certification_id == certification_id,
        )
    )).all()

    total = len(items)
    decided = sum(1 for it in items if it.decision != DECISION_PENDING)
    by_decision = {DECISION_PENDING: 0, DECISION_KEEP: 0, DECISION_REVOKE: 0}
    by_type: dict[str, int] = {}
    flagged: list[dict[str, Any]] = []
    pending_by_holder: dict[uuid.UUID, dict[str, Any]] = {}
    now = _naive_now()

    for it in items:
        by_decision[it.decision] = by_decision.get(it.decision, 0) + 1
        by_type[it.access_type] = by_type.get(it.access_type, 0) + 1
        if it.decision == DECISION_REVOKE:
            flagged.append({
                "id": str(it.id),
                "holder_email": it.holder_email,
                "holder_display_name": it.holder_display_name,
                "access_type": it.access_type,
                "access_label": it.access_label,
                "revocation_status": it.revocation_status,
                "decided_at": (
                    _to_naive(it.decided_at).isoformat()
                    if it.decided_at else None
                ),
            })
        elif it.decision == DECISION_PENDING and it.person_id is not None:
            overdue = bool(
                it.decided_at_deadline
                and _to_naive(it.decided_at_deadline) < now
            )
            agg = pending_by_holder.setdefault(it.person_id, {
                "person_id": str(it.person_id),
                "holder_email": it.holder_email,
                "holder_display_name": it.holder_display_name,
                "pending": 0,
                "overdue": False,
            })
            agg["pending"] += 1
            agg["overdue"] = agg["overdue"] or overdue

    # Disposition roll-up for the flagged-removal queue.
    disposition: dict[str, int] = {}
    for f in flagged:
        disposition[f["revocation_status"]] = (
            disposition.get(f["revocation_status"], 0) + 1
        )

    due = _to_naive(cert.due_at)
    return {
        "campaign": {
            "id": str(cert.id),
            "label": cert.label,
            "status": cert.status,
            "started_at": (
                _to_naive(cert.started_at).isoformat()
                if cert.started_at else None
            ),
            "due_at": due.isoformat() if due else None,
            "completed_at": (
                _to_naive(cert.completed_at).isoformat()
                if cert.completed_at else None
            ),
        },
        "total_items": total,
        "decided": decided,
        "coverage_pct": round(decided / total * 100, 1) if total else 0.0,
        "by_decision": by_decision,
        "by_type": by_type,
        "flagged_removals": sorted(
            flagged, key=lambda f: (f["revocation_status"], f["holder_email"] or "")
        ),
        "flagged_disposition": disposition,
        "non_responders": sorted(
            pending_by_holder.values(),
            key=lambda r: (not r["overdue"], r["holder_display_name"] or ""),
        ),
    }


async def build_team_attestation_status(
    db: AsyncSession, *, person_ids: list[uuid.UUID],
) -> dict[str, Any]:
    """Per-person attestation status for the CURRENT open campaign, scoped
    to the given people (the manager's reports). Read-only; the manager
    surface calls this with their subordinate ids. Returns empty when no
    campaign is open."""
    cert = await latest_open_self_attestation(db)
    if cert is None or not person_ids:
        return {"open": False, "campaign": None, "rows": []}

    rows = (await db.execute(
        select(
            AccessCertificationItem.person_id,
            AccessCertificationItem.holder_email,
            AccessCertificationItem.holder_display_name,
            AccessCertificationItem.decision,
            AccessCertificationItem.escalated_at,
        ).where(
            AccessCertificationItem.certification_id == cert.id,
            AccessCertificationItem.person_id.in_(person_ids),
        )
    )).all()

    by_person: dict[uuid.UUID, dict[str, Any]] = {}
    for pid, email, name, decision, escalated_at in rows:
        agg = by_person.setdefault(pid, {
            "person_id": str(pid),
            "holder_email": email,
            "holder_display_name": name,
            "total": 0, "pending": 0, "escalated": False,
        })
        agg["total"] += 1
        if decision == DECISION_PENDING:
            agg["pending"] += 1
        if escalated_at is not None:
            agg["escalated"] = True

    due = _to_naive(cert.due_at)
    return {
        "open": True,
        "campaign": {
            "id": str(cert.id), "label": cert.label,
            "due_at": due.isoformat() if due else None,
        },
        "rows": sorted(
            by_person.values(),
            key=lambda r: (r["pending"] == 0, r["holder_display_name"] or ""),
        ),
    }